Source code for repobee_testhelpers.localapi

"""A local implementation of the :py:class:`repobee_plug.PlatformAPI`
specification that can be used to test RepoBee and plugins.

.. danger::

    This module is in alpha version, and its behavior may change without

import pathlib
import pickle
import datetime
import dataclasses

from typing import List, Iterable, Optional, Set

import git  # type: ignore

import repobee_plug as plug


[docs]@dataclasses.dataclass(frozen=True) class User: username: str
[docs]@dataclasses.dataclass class Issue: title: str body: str number: int created_at: str author: str state: plug.IssueState assignees: Set[User] def to_plug_issue(self): return plug.Issue( title=self.title, body=self.body, number=self.number, created_at=self.created_at,, state=self.state, implementation=self, )
[docs]@dataclasses.dataclass(frozen=True) class Repo: name: str description: str url: str private: bool path: pathlib.Path issues: List[Issue] = dataclasses.field(default_factory=list) def to_plug_repo(self) -> plug.Repo: return plug.Repo(, description=self.description, private=self.private, url=self.url, implementation=self, ) def __hash__(self): return hash(
[docs]@dataclasses.dataclass(frozen=True) class Team: name: str members: Set[User] permission: plug.TeamPermission id: str repos: Set[Repo] = dataclasses.field(default_factory=set) def add_members(self, users: List[User]) -> None: for user in users: self.members.add(user) def to_plug_team(self) -> plug.Team: return plug.Team(, members=[mem.username for mem in self.members],, implementation=self, )
[docs]class LocalAPI(plug.PlatformAPI): """A local implementation of the :py:class:`repobee_plug.PlatformAPI` specification, which emulates a GitHub-like platform without accessing any network resources. """ def __init__(self, base_url: str, org_name: str, user: str, token: str): self._repodir = pathlib.Path(base_url[len("https://") :]) self._org_name = org_name self._user = user self._token = token self._teams: dict = {self._org_name: {}} self._repos: dict = {self._org_name: {}} self._users: dict = {} self._restore_state()
[docs] def create_team( self, name: str, members: Optional[List[str]] = None, permission: plug.TeamPermission = plug.TeamPermission.PUSH, ) -> plug.Team: """See :py:meth:`repobee_plug.PlatformAPI.create_team`.""" stored_team = self._teams[self._org_name].setdefault( name, Team(name=name, members=set(), permission=permission, id=name), ) self.assign_members( stored_team.to_plug_team(), members or [], permission ) return stored_team.to_plug_team()
[docs] def delete_team(self, team: plug.Team) -> None: """See :py:meth:`repobee_plug.PlatformAPI.delete_team`.""" del self._teams[self._org_name][]
[docs] def get_teams( self, team_names: Optional[Iterable[str]] = None ) -> Iterable[plug.Team]: """See :py:meth:`repobee_plug.PlatformAPI.get_teams`.""" team_names = set(team_names or []) return [ team.to_plug_team() for team in self._teams[self._org_name].values() if not team_names or in team_names ]
[docs] def assign_members( self, team: plug.Team, members: Iterable[str], permission: plug.TeamPermission = plug.TeamPermission.PUSH, ) -> None: """See :py:meth:`repobee_plug.PlatformAPI.assign_members`.""" team.implementation.add_members( [self._get_user(m) for m in members or []] )
[docs] def assign_repo( self, team: plug.Team, repo: plug.Repo, permission: plug.TeamPermission ) -> None: """See :py:meth:`repobee_plug.PlatformAPI.assign_repo`.""" team.implementation.repos.add(repo.implementation)
[docs] def create_repo( self, name: str, description: str, private: bool, team: Optional[plug.Team] = None, ) -> plug.Repo: """See :py:meth:`repobee_plug.PlatformAPI.create_repo`.""" assert not team or team.implementation repo_bucket = self._repos.setdefault(self._org_name, {}) if name in repo_bucket: raise plug.PlatformError(f"{name} already exists") repo_path = self._repodir / self._org_name / name repo_path.mkdir(parents=True, exist_ok=True) git.Repo.init(repo_path, bare=True) repo_bucket[name] = Repo( name=name, description=description, url=repo_path.as_uri(), private=private, path=repo_path, ) repo = repo_bucket[name] if team: self._get_team( return repo.to_plug_repo()
[docs] def get_repo(self, repo_name: str, team_name: Optional[str]) -> plug.Repo: """See :py:meth:`repobee_plug.PlatformAPI.get_repo`.""" repos = ( self._get_team(team_name).repos if team_name else self._repos[self._org_name].values() ) for repo in repos: if == repo_name: return repo.to_plug_repo() raise plug.NotFoundError(f"{team_name} has no repository {repo_name}")
[docs] def get_repos( self, repo_urls: Optional[List[str]] = None ) -> Iterable[plug.Repo]: """See :py:meth:`repobee_plug.PlatformAPI.get_repos`.""" repo_names = map(self.extract_repo_name, repo_urls or []) unfiltered_repos = ( (self._repos[self._org_name].get(name) for name in repo_names) if repo_urls else self._repos[self._org_name].values() ) return [repo.to_plug_repo() for repo in unfiltered_repos if repo]
[docs] def insert_auth(self, url: str) -> str: """See :py:meth:`repobee_plug.PlatformAPI.insert_auth`.""" if f"file://{self._repodir}" not in url: raise plug.InvalidURL(f"url not found on platform: '{url}'") return url
[docs] def create_issue( self, title: str, body: str, repo: plug.Repo, assignees: Optional[Iterable[str]] = None, ) -> plug.Issue: """See :py:meth:`repobee_plug.PlatformAPI.create_issue`.""" unique_assignees = { self._get_user(assignee) for assignee in (assignees or []) } issue = Issue( title=title, body=body, number=len(repo.implementation.issues), created_at=TIME, author=self._user, state=plug.IssueState.OPEN, assignees=unique_assignees, ) repo.implementation.issues.append(issue) return issue.to_plug_issue()
[docs] def close_issue(self, issue: plug.Issue) -> None: """See :py:meth:`repobee_plug.PlatformAPI.close_issue`.""" assert issue.implementation issue.implementation.state = plug.IssueState.CLOSED
[docs] def get_team_repos(self, team: plug.Team) -> Iterable[plug.Repo]: """See :py:meth:`repobee_plug.PlatformAPI.get_team_repos`.""" return (repo.to_plug_repo() for repo in team.implementation.repos)
[docs] def get_repo_issues(self, repo: plug.Repo) -> Iterable[plug.Issue]: """See :py:meth:`repobee_plug.PlatformAPI.get_repo_issues`.""" return (issue.to_plug_issue() for issue in repo.implementation.issues)
[docs] def get_repo_urls( self, assignment_names: Iterable[str], org_name: Optional[str] = None, team_names: Optional[List[str]] = None, insert_auth: bool = False, ) -> List[str]: assert not insert_auth, "not yet implemented" base = self._repodir / (org_name or self._org_name) repo_names = ( assignment_names if not team_names else plug.generate_repo_names(team_names, assignment_names) ) return [(base / name).as_uri() for name in repo_names]
[docs] def extract_repo_name(self, repo_url: str) -> str: return pathlib.Path(repo_url).stem
[docs] @staticmethod def verify_settings( user: str, org_name: str, base_url: str, token: str, template_org_name: Optional[str] = None, ) -> None: pass
def __getattribute__(self, key): attr = object.__getattribute__(self, key) if ( not key.startswith("_") and hasattr(plug.PlatformAPI, key) and callable(attr) ): # automatically save state after each API call def _func(*args, **kwargs): res = attr(*args, **kwargs) self._save_state() return res return _func return attr def _save_state(self): pickle_path = self._repodir / "state.pickle" pickle_path.write_bytes(pickle.dumps(self)) def _restore_state(self): pickle_path = self._repodir / "state.pickle" if pickle_path.is_file(): state = pickle.loads(pickle_path.read_bytes()) self.__dict__ = state.__dict__ def _add_users(self, usernames: List[str]) -> None: """Add users to this instance. .. note:: This function is public for use in testing. Args: usernames: A list of usernames to add. """ self._users.update({name: User(name) for name in usernames}) self._save_state() def _get_user(self, username: str) -> User: if username not in self._users: raise plug.NotFoundError(f"no such user: {username}") return self._users[username] def _get_team(self, team_id: str) -> Team: if team_id not in self._teams[self._org_name]: raise plug.NotFoundError(f"invalid team id: {team_id}") return self._teams[self._org_name][team_id]
[docs]class FakeAPIHooks(plug.Plugin): def api_init_requires(self): return ("base_url", "org_name", "user", "token") def get_api_class(self): return LocalAPI