Refactor Config()

webhook_server_container/app.py

@@ -1,12 +1,10 @@
-from typing import Any, Dict
 import os
 import sys
+from typing import Any
 
-from fastapi import Request
 import requests
 import urllib3
-
-from fastapi import FastAPI
+from fastapi import FastAPI, Request
 
 from webhook_server_container.libs.github_api import ProcessGithubWehook
 from webhook_server_container.utils.helpers import get_logger_with_params
@@ -17,22 +15,22 @@
 
 
 @FASTAPI_APP.get(f"{APP_URL_ROOT_PATH}/healthcheck")
-def healthcheck() -> Dict[str, Any]:
+def healthcheck() -> dict[str, Any]:
     return {"status": requests.codes.ok, "message": "Alive"}
 
 
 @FASTAPI_APP.post(APP_URL_ROOT_PATH)
-async def process_webhook(request: Request) -> Dict[str, Any]:
+async def process_webhook(request: Request) -> dict[str, Any]:
     logger_name: str = "main"
     logger = get_logger_with_params(name=logger_name)
     delivery_headers = request.headers.get("X-GitHub-Delivery", "")
-    process_failed_msg: Dict[str, Any] = {
+    process_failed_msg: dict[str, Any] = {
         "status": requests.codes.server_error,
         "message": "Process failed",
         "log_prefix": delivery_headers,
     }
     try:
-        hook_data: Dict[Any, Any] = await request.json()
+        hook_data: dict[Any, Any] = await request.json()
 
     except Exception as ex:
         logger.error(f"Error get JSON from request: {ex}")

webhook_server_container/libs/config.py

@@ -1,23 +1,36 @@
 import os
-from typing import Any, Dict
+from typing import Any
 
 import yaml
 
 
 class Config:
-    def __init__(self) -> None:
+    def __init__(self, repository: str | None = None) -> None:
         self.data_dir: str = os.environ.get("WEBHOOK_SERVER_DATA_DIR", "/home/podman/data")
         self.config_path: str = os.path.join(self.data_dir, "config.yaml")
         self.exists()
+        self.repository = repository
 
     def exists(self) -> None:
         if not os.path.isfile(self.config_path):
             raise FileNotFoundError(f"Config file {self.config_path} not found")
 
     @property
-    def data(self) -> Dict[str, Any]:
+    def data(self) -> dict[str, Any]:
         with open(self.config_path) as fd:
             return yaml.safe_load(fd)
 
-    def repository_data(self, repository_name: str) -> Dict[str, Any]:
-        return self.data["repositories"].get(repository_name, {})
+    @property
+    def repository_data(self) -> dict[str, Any]:
+        return self.data["repositories"].get(self.repository, {})
+
+    def get_value(self, value: str, return_on_none: Any = None) -> Any:
+        """
+        Get value from config, try first from repository and if not exists get it from root config
+        """
+        _val = self.repository_data.get(value)
+
+        if _val is None:
+            _val = self.data.get(value)
+
+        return _val or return_on_none

webhook_server_container/libs/jira_api.py

@@ -1,5 +1,6 @@
-from typing import Any, Dict, List
-from jira import Issue, JIRA
+from typing import Any
+
+from jira import JIRA, Issue
 
 from webhook_server_container.utils.helpers import get_logger_with_params
 
@@ -16,7 +17,7 @@ def __init__(self, server: str, project: str, token: str):
             token_auth=self.token,
         )
         self.conn.my_permissions()
-        self.fields: Dict[str, Any] = {"project": {"key": self.project}}
+        self.fields: dict[str, Any] = {"project": {"key": self.project}}
 
     def create_story(self, title: str, body: str, epic_key: str, assignee: str) -> str:
         self.fields.update({
@@ -51,5 +52,5 @@ def close_issue(self, key: str, comment: str = "") -> None:
         )
 
     def get_epic_custom_field(self) -> str:
-        _epic_field_id: List[str] = [cf["id"] for cf in self.conn.fields() if "Epic Link" in cf["name"]]
+        _epic_field_id: list[str] = [cf["id"] for cf in self.conn.fields() if "Epic Link" in cf["name"]]
         return _epic_field_id[0] if _epic_field_id else ""

webhook_server_container/tests/conftest.py

@@ -49,6 +49,7 @@ def decoded_content(self):
 class Repository:
     def __init__(self):
         self.name = "test-repo"
+        self.full_name = "my-org/test-repo"
 
     def get_git_tree(self, sha: str, recursive: bool):
         return Tree("")
@@ -104,7 +105,9 @@ def process_github_webhook(mocker, request):
     mocker.patch(f"{base_import_path}.get_github_repo_api", return_value=Repository())
 
     process_github_webhook = ProcessGithubWehook(
-        {"repository": {"name": Repository().name}}, Headers({"X-GitHub-Event": "test-event"}), logging.getLogger()
+        hook_data={"repository": {"name": Repository().name, "full_name": Repository().full_name}},
+        headers=Headers({"X-GitHub-Event": "test-event"}),
+        logger=logging.getLogger(),
     )
     process_github_webhook.pull_request_branch = "main"
     if hasattr(request, "param") and request.param:

webhook_server_container/tests/manifests/config.yaml

@@ -3,14 +3,14 @@ log-file: webhook-server.log # Set global log file, change take effect immediate
 
 github-app-id: 123456 # GitHub app id
 github-toekns:
-  - <GITHIB TOKEN1>
-  - <GITHIB TOKEN2>
+  - GITHIB TOKEN1
+  - GITHIB TOKEN2
 
-webhook_ip: <HTTP://IP OR URL:PORT>
+webhook_ip: HTTP://IP OR URL:PORT
 
 docker: # Used to pull images from docker.io
-  username: <username>
-  password: <password>
+  username: username
+  password: password # pragma: allowlist secret
 
 default-status-checks:
   - "WIP"
@@ -22,21 +22,21 @@ auto-verified-and-merged-users:
   - "pre-commit-ci[bot]"
 
 jira:
-  server: <JIRA URL>
-  project: <PROJECT KEY>
-  token: <JIRA TOKEN>
+  server: JIRA URL
+  project: PROJECT KEY
+  token: JIRA TOKEN
   user-mapping:
-    <GITHUB USER>: <JIRA USER> # if github user is not the same as jira
+    GITHUB USER: <JIRA USER> # if github user is not the same as jira
 
 repositories:
   test-repo:
     name: my-org/test-repo
     log-level: DEBUG # Override global log-level for repository
     log-file: test-repo.log # Override global log-file for repository
-    slack_webhook_url: <Slack webhook url> # Send notification to slack on several operations
+    slack_webhook_url: Slack webhook url # Send notification to slack on several operations
     verified_job: true
     pypi:
-      token: <PYPI TOKEN>
+      token: PYPI TOKEN
 
     events: # To listen to all events do not send events
       - push
@@ -60,10 +60,10 @@ repositories:
         exclude-runs:
           - "SonarCloud Code Analysis"
     container:
-      username: <registry username>
-      password: <registry_password>
-      repository: <registry_repository_full_path>
-      tag: <image_tag>
+      username: registry username
+      password: registry_password # pragma: allowlist secret
+      repository: registry_repository_full_path
+      tag: image_tag
       release: true # Push image to registry on new release with release as the tag
       build-args: # build args to send to podman build command
         - my-build-arg1=1
@@ -75,8 +75,8 @@ repositories:
       - "my[bot]"
 
     github-tokens: # override GitHub tokens per repository
-      - <GITHUB TOKEN1>
-      - <GITHUB TOKEN2>
+      - GITHUB TOKEN1
+      - GITHUB TOKEN2
 
     can-be-merged-required-labels: # check for extra labels to set PR as can be merged
       - my-label1
@@ -85,9 +85,9 @@ repositories:
     jira-tracking: true
 
     jira: # override Jira global settings
-      server: <JIRA URL>
-      project: <PROJECT KEY>
-      token: <JIRA TOKEN>
-      epic: <EPIC KEY> # Optional
+      server: JIRA URL
+      project: PROJECT KEY
+      token: JIRA TOKEN
+      epic: EPIC KEY # Optional
       user-mapping:
-        <GITHUB USER>: <JIRA USER> # if github user is not the same as jira
+        GITHUB USER: <JIRA USER> # if github user is not the same as jira

webhook_server_container/tests/test_branch_protection.py

@@ -1,26 +1,30 @@
 import os
+
 import pytest
 from webhook_server_container.libs.config import Config
 from webhook_server_container.utils.github_repository_settings import (
-    get_repo_branch_protection_rules,
     DEFAULT_BRANCH_PROTECTION,
+    get_repo_branch_protection_rules,
 )
 
 
 @pytest.fixture()
 def branch_protection_rules(request, mocker):
     os.environ["WEBHOOK_SERVER_DATA_DIR"] = "webhook_server_container/tests/manifests"
-    config = Config()
     repo_name = "test-repo"
+    config = Config(repository=repo_name)
     data = config.data
     data.setdefault("branch_protection", request.param.get("global", {}))
-    data["repositories"][repo_name].setdefault("branch_protection", request.param.get("repo", {}))
+    data["repositories"][repo_name].setdefault("branch_protection", request.param.get("repo"))
     mocker.patch(
         "webhook_server_container.libs.config.Config.data", new_callable=mocker.PropertyMock, return_value=data
     )
-    return get_repo_branch_protection_rules(config_data=config.data, repo_data=config.data["repositories"][repo_name])[
-        "branch_protection"
-    ]
+    mocker.patch(
+        "webhook_server_container.libs.config.Config.repository_data",
+        new_callable=mocker.PropertyMock,
+        return_value=data["repositories"][repo_name],
+    )
+    return get_repo_branch_protection_rules(config=config)
 
 
 @pytest.mark.parametrize(
@@ -92,4 +96,5 @@ def test_branch_protection_setup(branch_protection_rules, expected):
     for key in expected:
         if branch_protection_rules[key] != expected[key]:
             mismatch[key] = f"Expected value for {key}: {expected[key]}, actual: {branch_protection_rules[key]}"
+
     assert not mismatch, f"Following mismatches are found: {mismatch}"

webhook_server_container/tests/test_repo_data_from_config.py

@@ -3,15 +3,15 @@ def test_repo_data_from_config_repository_found(process_github_webhook):
 
     assert process_github_webhook.repository_full_name == "my-org/test-repo"
     assert process_github_webhook.github_app_id == 123456
-    assert process_github_webhook.pypi == {"token": "<PYPI TOKEN>"}
+    assert process_github_webhook.pypi == {"token": "PYPI TOKEN"}
     assert process_github_webhook.verified_job
     assert process_github_webhook.tox_python_version == "3.8"
-    assert process_github_webhook.slack_webhook_url == "<Slack webhook url>"
-    assert process_github_webhook.container_repository_username == "<registry username>"
-    assert process_github_webhook.container_repository_password == "<registry_password>"
-    assert process_github_webhook.container_repository == "<registry_repository_full_path>"
+    assert process_github_webhook.slack_webhook_url == "Slack webhook url"
+    assert process_github_webhook.container_repository_username == "registry username"
+    assert process_github_webhook.container_repository_password == "registry_password"  # pragma: allowlist secret
+    assert process_github_webhook.container_repository == "registry_repository_full_path"
     assert process_github_webhook.dockerfile == "Dockerfile"
-    assert process_github_webhook.container_tag == "<image_tag>"
+    assert process_github_webhook.container_tag == "image_tag"
     assert process_github_webhook.container_build_args == ["my-build-arg1=1", "my-build-arg2=2"]
     assert process_github_webhook.container_command_args == ["--format docker"]
     assert process_github_webhook.container_release

webhook_server_container/utils/constants.py

@@ -1,5 +1,3 @@
-from typing import Dict
-
 OTHER_MAIN_BRANCH: str = "master"
 TOX_STR: str = "tox"
 PRE_COMMIT_STR: str = "pre-commit"
@@ -36,14 +34,14 @@
 COMMAND_ASSIGN_REVIEWER_STR = "assign-reviewer"
 
 # Gitlab colors require a '#' prefix; e.g: #
-USER_LABELS_DICT: Dict[str, str] = {
+USER_LABELS_DICT: dict[str, str] = {
     HOLD_LABEL_STR: "B60205",
     VERIFIED_LABEL_STR: "0E8A16",
     WIP_STR: "B60205",
     LGTM_STR: "0E8A16",
 }
 
-STATIC_LABELS_DICT: Dict[str, str] = {
+STATIC_LABELS_DICT: dict[str, str] = {
     **USER_LABELS_DICT,
     CHERRY_PICKED_LABEL_PREFIX: "1D76DB",
     f"{SIZE_LABEL_PREFIX}L": "F5621C",
@@ -57,7 +55,7 @@
     HAS_CONFLICTS_LABEL_STR: "B60205",
 }
 
-DYNAMIC_LABELS_DICT: Dict[str, str] = {
+DYNAMIC_LABELS_DICT: dict[str, str] = {
     APPROVED_BY_LABEL_PREFIX: "0E8A16",
     LGTM_BY_LABEL_PREFIX: "DCED6F",
     COMMENTED_BY_LABEL_PREFIX: "D93F0B",
@@ -67,7 +65,7 @@
     JIRA_STR: "1D76DB",
 }
 
-ALL_LABELS_DICT: Dict[str, str] = {**STATIC_LABELS_DICT, **DYNAMIC_LABELS_DICT}
+ALL_LABELS_DICT: dict[str, str] = {**STATIC_LABELS_DICT, **DYNAMIC_LABELS_DICT}
 
 
 class REACTIONS: