Unit test for token usage (#107)

Author: PeeachPie

.github/workflows/tests.yml

@@ -51,3 +51,23 @@ jobs:
 
     - name: Show out.csv
       run: cat out.csv
+
+  token-usage-unit_test:
+
+      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: Run test
+        run: python3 test_token_usage.py --tt1 ${{ secrets.TEST_TOKEN_GITHUB }} --tt2 ${{ secrets.SECOND_TEST_TOKEN_GITHUB }} --repo moevm/github_repo_commitment_calc --out out.csv

ForgejoRepoAPI.py

@@ -4,9 +4,20 @@
 import isodate
 from pyforgejo import PyforgejoApi
 
-from interface_wrapper import (Branch, Comment, Commit, Contributor, Invite,
-                               IRepositoryAPI, Issue, PullRequest, Repository,
-                               User, WikiPage, logging)
+from interface_wrapper import (
+    Branch,
+    Comment,
+    Commit,
+    Contributor,
+    Invite,
+    IRepositoryAPI,
+    Issue,
+    PullRequest,
+    Repository,
+    User,
+    WikiPage,
+    logging,
+)
 
 
 class ForgejoRepoAPI(IRepositoryAPI):

GitHubRepoAPI.py

@@ -1,12 +1,37 @@
-from interface_wrapper import (Branch, Comment, Commit, Contributor, Invite,
-                               IRepositoryAPI, Issue, PullRequest, Repository,
-                               User, WikiPage, logging)
+from interface_wrapper import (
+    Branch,
+    Comment,
+    Commit,
+    Contributor,
+    Invite,
+    IRepositoryAPI,
+    Issue,
+    PullRequest,
+    Repository,
+    User,
+    WikiPage,
+    logging,
+)
+
+from github import Github, GithubException
 
 
 class GitHubRepoAPI(IRepositoryAPI):
+    def __init__(self, client: Github):
+        self.client = self._client_validation(client)
 
-    def __init__(self, client):
-        self.client = client
+    @staticmethod
+    def _client_validation(client: Github) -> Github:
+        try:
+            client.get_user().login
+        except GithubException as err:
+            logging.error(f'Github: Connect: error {err.data}')
+            logging.error(
+                'Github: Connect: user could not be authenticated please try again.'
+            )
+            exit(1)
+        else:
+            return client
 
     def get_user_data(self, user) -> User:
         return User(

commits_parser.py

@@ -1,5 +1,7 @@
 from dataclasses import asdict, dataclass
 from time import sleep
+from typing import Generator
+from datetime import datetime
 
 import pytz
 
@@ -62,12 +64,17 @@ def log_repository_commits(
 
 
 def log_commits(
-    client: IRepositoryAPI, working_repos, csv_name, start, finish, branch, fork_flag
+    binded_repos: Generator[tuple[IRepositoryAPI, Repository, str], None, None],
+    csv_name: str,
+    start: datetime,
+    finish: datetime,
+    branch: str,
+    fork_flag: bool,
 ):
     info = asdict(CommitData())
     logger.log_to_csv(csv_name, list(info.keys()))
 
-    for repo, token in working_repos:
+    for client, repo, token in binded_repos:
         try:
             logger.log_title(repo.name)
             log_repository_commits(client, repo, csv_name, start, finish, branch)

contributors_parser.py

@@ -80,12 +80,14 @@ def get_contributors_stats(client: IRepositoryAPI, repository: Repository) -> di
 
 
 def log_contributors(
-    client: IRepositoryAPI, working_repos: Generator, csv_name: str, fork_flag: bool
+    binded_repos: Generator[tuple[IRepositoryAPI, Repository, str], None, None],
+    csv_name: str,
+    fork_flag: bool,
 ):
     info = asdict(ContributorData())
     logger.log_to_csv(csv_name, list(info.keys()))
 
-    for repo, token in working_repos:
+    for client, repo, token in binded_repos:
         try:
             logger.log_title(repo.name)
             log_repository_contributors(client, repo, csv_name)

git_logger.py

@@ -16,22 +16,40 @@ def get_tokens_from_file(tokens_path: str) -> list[str]:
     return tokens
 
 
-class GitClients:
+def get_repos_from_file(repos_path: str) -> list[str]:
+    with open(repos_path, 'r') as file:
+        list_repos = [x for x in file.read().split('\n') if x]
+
+    return list_repos
+
+
+class Clients:
     def __init__(self, source: str, tokens: list[str], base_url: str | None = None):
-        self.clients = self._init_clients(source, tokens, base_url)
-        self.cur_client = None
+        # Возможно это можно переписать покрасивее
+        if source == 'github':
+            self.clients = self._init_clients(source, tokens, base_url)
+        elif base_url == 'forgejo':
+            self.client = RepositoryFactory.create_api(source, tokens[0], base_url)
+            self.token = tokens[0]
+        else:
+            print(f"Unavailable source {source}, use [ 'github' | 'forgejo' ] instead")
+
+        self.source = source
 
     def _init_clients(
         self, source: str, tokens: list[str], base_url: str | None
     ) -> list[dict]:
         clients = [
-            {"client": login(source, token, base_url), "token": token}
+            {
+                "client": RepositoryFactory.create_api(source, token, base_url),
+                "token": token,
+            }
             for token in tokens
         ]
 
         return clients
 
-    def get_next_client(self) -> IRepositoryAPI:
+    def _get_next_git_client(self) -> tuple[IRepositoryAPI, str]:
         client = None
         max_remaining_limit = -1
 
@@ -51,24 +69,29 @@ def get_next_client(self) -> IRepositoryAPI:
         if client is None:
             raise Exception("No git clients available")
 
-        self.cur_client = client
-        return client
+        return client['client'], client['token']
 
+    def _get_next_forgejo_client(self) -> tuple[IRepositoryAPI, str]:
+        return self.client, self.token
 
-def get_next_repo(clients: GitClients, repositories):
-    with open(repositories, 'r') as file:
-        list_repos = [x for x in file.read().split('\n') if x]
-    print(list_repos)
-    for repo_name in list_repos:
+    def get_next_client(self) -> tuple[IRepositoryAPI, str]:
+        if self.source == 'github':
+            return self._get_next_git_client()
+        elif self.source == 'forgejo':
+            return self._get_next_forgejo_client
+
+
+def get_next_binded_repo(clients: Clients, repositories: list[str]):
+    for repo_name in repositories:
         try:
-            cur_client = clients.get_next_client()
-            repo = cur_client['client'].get_repository(repo_name)
+            client, token = clients.get_next_client()
+            repo = client.get_repository(repo_name)
         except Exception as err:
             print(f'get_next_repo(): error {err}')
             print(f'get_next_repo(): failed to load repository "{repo_name}"')
             exit(1)
         else:
-            yield repo, cur_client['token']
+            yield client, repo, token
 
 
 def get_assignee_story(git_object):

interface_wrapper.py

@@ -3,7 +3,7 @@
 from dataclasses import dataclass
 from datetime import datetime
 
-from github import Github
+from github import Github, Auth
 from pyforgejo import PyforgejoApi
 
 # Настройка логирования
@@ -182,7 +182,7 @@ def create_api(
         from GitHubRepoAPI import GitHubRepoAPI
 
         if source == 'github':
-            return GitHubRepoAPI(Github(token))
+            return GitHubRepoAPI(Github(auth=Auth.Token(token)))
         elif source == 'forgejo':
             if not isinstance(base_url, str):
                 raise ValueError(

invites_parser.py

@@ -1,5 +1,6 @@
 from dataclasses import asdict, dataclass
 from time import sleep
+from typing import Generator
 
 from constants import TIMEDELTA
 from interface_wrapper import IRepositoryAPI, Repository
@@ -31,11 +32,14 @@ def log_repository_invitations(
         sleep(TIMEDELTA)
 
 
-def log_invitations(client: IRepositoryAPI, working_repos, csv_name: str):
+def log_invitations(
+    binded_repos: Generator[tuple[IRepositoryAPI, Repository, str], None, None],
+    csv_name: str,
+):
     info = asdict(InviteData())
     logger.log_to_csv(csv_name, list(info.keys()))
 
-    for repo, token in working_repos:
+    for client, repo, token in binded_repos:
         logger.log_title(repo.name)
         try:
             log_repository_invitations(client, repo, csv_name)

issues_parser.py

@@ -1,6 +1,8 @@
 import json
 from dataclasses import asdict, dataclass
 from time import sleep
+from typing import Generator
+from datetime import datetime
 
 import pytz
 import requests
@@ -187,12 +189,16 @@ def log_issue_and_comments(csv_name, issue_data: IssueData, comments):
 
 
 def log_issues(
-    client: IRepositoryAPI, working_repo, csv_name, token, start, finish, fork_flag
+    binded_repos: Generator[tuple[IRepositoryAPI, Repository, str], None, None],
+    csv_name: str,
+    start: datetime,
+    finish: datetime,
+    fork_flag: bool,
 ):
     info = asdict(IssueDataWithComment())
     logger.log_to_csv(csv_name, list(info.keys()))
 
-    for repo, token in working_repo:
+    for client, repo, token in binded_repos:
         try:
             logger.log_title(repo.name)
             log_repository_issues(client, repo, csv_name, token, start, finish)