76 lines
2.1 KiB
Python
76 lines
2.1 KiB
Python
"""Shared fixtures for the Atlas-only Gitea client contracts."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import json
|
|
import sys
|
|
from email.message import Message
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).parents[2]
|
|
CLIENT_PATH = ROOT / "services/hermes/scm-common/scripts/gitea_api.py"
|
|
HEAD_SHA = "465cf9146b05c174a2a8d310aff6c64be58277b6"
|
|
if str(CLIENT_PATH.parent) not in sys.path:
|
|
sys.path.insert(0, str(CLIENT_PATH.parent))
|
|
|
|
|
|
def _load():
|
|
spec = importlib.util.spec_from_file_location("safe_gitea_api", CLIENT_PATH)
|
|
assert spec and spec.loader
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[spec.name] = module
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def _draft_payload(**updates):
|
|
payload = {
|
|
"base": "main",
|
|
"body": "Review evidence",
|
|
"head": "hermes/review-fix",
|
|
"title": "WIP: Repair review findings",
|
|
}
|
|
payload.update(updates)
|
|
return payload
|
|
|
|
|
|
def _draft_response(**updates):
|
|
response = {
|
|
"number": 3,
|
|
"state": "open",
|
|
"draft": True,
|
|
"merged": False,
|
|
"html_url": "https://scm.bstein.dev/atlas/cassandra/pulls/3",
|
|
"url": "https://scm.bstein.dev/atlas/cassandra/pulls/3",
|
|
"title": "WIP: Focused fix",
|
|
"body": "Review evidence",
|
|
"base": {"ref": "main", "repo": {"full_name": "atlas/cassandra"}},
|
|
"head": {
|
|
"ref": "hermes/fix",
|
|
"sha": HEAD_SHA,
|
|
"repo": {"full_name": "atlas/cassandra"},
|
|
},
|
|
}
|
|
response.update(updates)
|
|
return response
|
|
|
|
|
|
class Response:
|
|
def __init__(self, body: object, status: int | None = None):
|
|
self.body = body if isinstance(body, bytes) else json.dumps(body).encode()
|
|
self.status = (
|
|
status if status is not None else (201 if isinstance(body, dict) else 200)
|
|
)
|
|
self.headers = Message()
|
|
self.headers["Content-Type"] = "application/json"
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_args):
|
|
return False
|
|
|
|
def read(self, limit=-1):
|
|
return self.body if limit < 0 else self.body[:limit]
|