version-control-platform/vcp_management/models/vcp_repository_branch.py at 3520a5a6f2f552ef9fd73f520061a5990288cb37 · grap/version-control-platform · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
# Copyright 2026 Dixmit
# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl).
import os
import re

import git

from odoo import _, api, fields, models
from odoo.exceptions import ValidationError


class VcpRepositoryBranch(models.Model):
    _name = "vcp.repository.branch"
    _inherit = ["vcp.rule.information.mixin"]
    _description = "Links Branches with Repositories"

    branch_id = fields.Many2one(
        "vcp.branch",
        string="Branch",
        required=True,
        readonly=True,
        ondelete="cascade",
    )
    repository_id = fields.Many2one(
        "vcp.repository",
        required=True,
        readonly=True,
        ondelete="cascade",
    )
    platform_id = fields.Many2one(
        related="repository_id.platform_id",
        readonly=True,
    )
    last_commit = fields.Char(readonly=True)
    rule_ids = fields.Many2many(
        "vcp.rule",
        string="Processing Rules",
    )
    override_parent_rules = fields.Boolean()
    update_rule_processing_date = fields.Datetime(
        default=fields.Datetime.now,
        required=True,
    )

    def _cron_process_branch_rules(self, limit):
        branches = self.search([], limit=limit, order="update_rule_processing_date asc")
        for branch in branches:
            branch.process_rules()

    @api.constrains("branch_id", "repository_id")
    def _check_branch_repository(self):
        for record in self:
            if record.branch_id.platform_id != record.repository_id.platform_id:
                raise ValidationError(
                    _("The branch and the repository must belong to the same platform.")
                )

    def _get_rules(self):
        rules = self.rule_ids
        if not self.override_parent_rules:
            rules |= self.repository_id._get_rules()
        return rules

    def _get_local_path(self):
        return f"{self.repository_id.local_path}/{self.branch_id.name}"

    def process_rules(self):
        for record in self:
            rules = record._get_rules()
            for rule in rules:
                if re.match(rule.branch_pattern, record.branch_id.name):
                    rule._process_rule(record)

    def _download_code(self):
        result = super()._download_code()
        local_path = self.local_path
        try:
            os.makedirs(local_path, exist_ok=True)
        except PermissionError as err:
            raise ValidationError(
                _(
                    "Unable to create a folder in '%(local_path)s'.",
                    local_path=local_path,
                )
            ) from err

        try:
            repo = git.Repo(local_path)
            for remote in repo.remotes:
                if remote.url == self.repository_id._get_git_url():
                    remote.fetch(self.branch_id.name)
                    repo.git.reset("--hard", f"{remote.name}/{self.branch_id.name}")
                    break
        except git.exc.InvalidGitRepositoryError:
            # Not cloned yet
            repo = git.Repo.clone_from(
                self.repository_id._get_git_url(),
                local_path,
                branch=self.branch_id.name,
                depth=1,
            )
        return result

    def _compute_display_name(self):
        if not self._context.get("display_only_branch_name"):
            return super()._compute_display_name()

        for record in self:
            record.display_name = record.branch_id.name