Fix forgejo base url

.flake8

@@ -1,4 +1,4 @@
 [flake8]
 max-line-length = 120
-exclude = .git,__pycache__,docs
+exclude = .git,__pycache__,docs,venv
 statistics = True

.github/workflows/smoke-tests-forgejo.yml

@@ -0,0 +1,95 @@
+name: '[Forgejo] Smoke Tests'
+
+on:
+  push:
+    branches: [ "master" ]
+  pull_request:
+    branches: [ "master" ]
+
+jobs:
+  smoke-test:
+    strategy:
+      fail-fast: false
+      matrix:
+        args:
+          - "--invites"
+          - "--commits"
+          - "--pull_requests"
+          - "--issues"
+          - "--wikis"
+          - "--contributors"
+          - "--workflow_runs"
+
+    runs-on: ubuntu-latest
+
+    steps:
+    - uses: actions/checkout@v4
+
+    - name: Cache pip
+      uses: actions/cache@v3
+      with:
+        path: ~/.cache/pip
+        key: ${{ runner.os }}-pip-${{ hashFiles('requirements.txt') }}
+        restore-keys: |
+          ${{ runner.os }}-pip-
+
+    - name: Install dependencies
+      run: pip install -r requirements.txt
+
+    - name: Create list.txt
+      run: echo "harabat/pyforgejo" > list.txt
+
+    - name: Run test
+      run: |
+        python3 main.py ${{ matrix.args }} --token ${{ secrets.TEST_TOKEN_FORGEJO }} --base_url https://codeberg.org/api/v1 --list list.txt --out out.csv --branch master
+
+    - name: Check if out.csv exists
+      run: ls out.csv
+
+    - name: Fail if out.csv does not exist
+      if: failure()
+      run: exit 1
+
+    - name: Show out.csv
+      run: cat out.csv
+
+    - name: Check header in first line
+      run: |
+        case "${{ matrix.args }}" in
+          --invites)
+            HEADER="repository name,invited login,invite creation date,invitation url"
+            ;;
+          --commits)
+            HEADER="repository name,author name,author login,author email,date and time,changed files,commit id,branch"
+            ;;
+          --pull_requests)
+            HEADER="repository name,title,id,state,commit into,commit from,created at,creator name,creator login,creator email,changed files,comment body,comment created at,comment author name,comment author login,comment author email,merger name,merger login,merger email,source branch,target branch,assignee story,related issues,labels,milestone"
+            ;;
+          --issues)
+            HEADER="repository name,number,title,state,task,created at,creator name,creator login,creator email,closer name,closer login,closer email,closed at,comment body,comment created at,comment author name,comment author login,comment author email,assignee story,connected pull requests,labels,milestone"
+            ;;
+          --wikis)
+            HEADER="repository name,author name,author login,datetime,page,action,revision id,added lines,deleted lines"
+            ;;
+          --contributors)
+            HEADER="repository name,login,name,email,url,permissions,total commits,node id,type,bio,site admin"
+            ;;
+          --workflow_runs)
+            HEADER="repository name,display title,event,head branch,head sha,name,path,started at,total duration,conclusion,status,url"
+            ;;
+          *)
+            echo "Unknown ARG: '${{ matrix.args }}'"
+            exit 1
+            ;;
+        esac
+
+        FIRST_LINE=$(head -n 1 out.csv)
+
+        if [[ "$FIRST_LINE" == "$HEADER"* ]]; then
+          echo "Header is valid for ${{ matrix.args }}"
+        else
+          echo "::error::Header is invalid for ${{ matrix.args }}"
+          echo "Expected: $HEADER"
+          echo "Actual:   $FIRST_LINE"
+          exit 1
+        fi

.github/workflows/smoke-tests.yml

@@ -1,4 +1,4 @@
-name: Smoke Tests
+name: '[Github] Smoke Tests'
 
 on:
   push:

src/ForgejoRepoAPI.py

@@ -275,7 +275,7 @@ def get_workflow_runs(self, repo) -> list[WorkflowRun]:
         return []
 
     def get_base_url(self) -> str:
-        return self.client.base_url
+        return self.client._client_wrapper.get_base_url()
 
 
 # Точка входа для тестирования

src/git_logger.py

@@ -88,39 +88,36 @@ def get_next_binded_repo(clients: Clients, repositories: list[str]):
 def get_assignee_story(git_object, client, token, repository):
     assignee_result = ""
 
-    try:
-        repo_owner = repository.owner.login
-        repo_name = repository.name
-        issue_index = git_object._id  # Для pull request и issue одинаково
-
-        base_url = client.get_base_url().rstrip('/')
-
-        url = f"{base_url}/repos/{repo_owner}/{repo_name}/issues/{issue_index}/timeline"
-        headers = {
-            "Authorization": f"Bearer {token}" if client is GitHubRepoAPI else f"token {token}",
-            "Accept": "application/json"
-        }
-
-        response = requests.get(url, headers=headers)
-        if response.status_code != 200:
-            raise Exception(f"Failed to fetch issue timeline: {response.status_code}, {response.text}")
-
-        events = response.json()
-
-        for event in events:
-            if event.get('event') in ["assigned", "unassigned"]:
-                date = event.get('created_at')
-                assigner = event.get('actor', {}).get('login', 'unknown')
-                assignee = event.get('assignee', {}).get('login', 'unknown')
-
-                assignee_result += f"{date}: {assigner} -"
-                if event['event'] == "unassigned":
-                    assignee_result += "/"
-                assignee_result += f"> {assignee}; "
-
-                sleep(TIMEDELTA)
-
-    except Exception as e:
-        print(f"get_assignee_story(): error {e}")
+    repo_owner = repository.owner.login
+    repo_name = repository.name
+    # Для pull request и issue одинаково
+    issue_index = (
+        getattr(git_object, "number", None)
+        or getattr(git_object, "index", None)
+        or getattr(git_object, "_id", None)
+    )
+
+    base_url = client.get_base_url().rstrip('/')
+
+    url = f"{base_url}/repos/{repo_owner}/{repo_name}/issues/{issue_index}/timeline"
+    headers = {
+        "Authorization": f"Bearer {token}" if client is GitHubRepoAPI else f"token {token}",
+        "Accept": "application/json"
+    }
+
+    response = requests.get(url, headers=headers)
+    if response.status_code != 200:
+        raise Exception(f"Failed to fetch issue timeline: {response.status_code}, {response.text}")
+
+    events = response.json()
+
+    results = [
+        f"{event.get('created_at')}: {event.get('actor', {}).get('login', 'unknown')} -"
+        + ("/" if event.get('event') == "unassigned" else "")
+        + f"> {event.get('assignee', {}).get('login', 'unknown')}; "
+        for event in events
+        if event.get('event') in ["assigned", "unassigned"]
+    ]
+    assignee_result = ''.join(results)
 
     return assignee_result