mjalmoudy94
717e53a7ce
مخزن اختصاصی برنامهٔ وب؛ محتوای پوشهٔ web در ریشهٔ مخزن قرار گرفت. تاریخچهٔ پروژهٔ ربات (agy-bot) در این مخزن وجود ندارد و هیچ وابستگی runtime به آن نیست.
118 lines
4.9 KiB
Python
118 lines
4.9 KiB
Python
"""Project Git repository endpoints.
|
|
|
|
These endpoints had **no test coverage** and were broken at runtime: the domain
|
|
entity and the infrastructure repository that manages ``Project`` rows were both
|
|
named ``ProjectRepository``, so the infrastructure class shadowed the entity and
|
|
``ensure_repository`` constructed it with ``project_id=`` — a ``TypeError`` that
|
|
turned every call into a 500.
|
|
|
|
The tests below exist to keep that from regressing, and to pin the security
|
|
properties the module promises: repositories are always private, and the Git
|
|
password never reaches a response body.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
from httpx import AsyncClient
|
|
|
|
from tests.conftest import auth_header, signup
|
|
|
|
|
|
async def _org_and_project(client: AsyncClient, token: str) -> tuple[str, str]:
|
|
org = await client.post(
|
|
"/api/organizations", json={"name": "شرکت مخزن"}, headers=auth_header(token)
|
|
)
|
|
assert org.status_code == 201, org.text
|
|
org_id = org.json()["id"]
|
|
|
|
project = await client.post(
|
|
f"/api/organizations/{org_id}/projects",
|
|
json={"name": "پروژهٔ آزمایشی", "description": "برای تست مخزن"},
|
|
headers=auth_header(token),
|
|
)
|
|
assert project.status_code == 201, project.text
|
|
return org_id, project.json()["id"]
|
|
|
|
|
|
async def test_ensure_repository_creates_a_private_repo(client: AsyncClient):
|
|
"""The exact path that used to raise TypeError -> 500."""
|
|
token = (await signup(client, "repo-owner@example.com"))["access_token"]
|
|
org_id, project_id = await _org_and_project(client, token)
|
|
base = f"/api/organizations/{org_id}/projects/{project_id}"
|
|
|
|
response = await client.post(f"{base}/repo/ensure", headers=auth_header(token))
|
|
assert response.status_code in (200, 201), response.text
|
|
|
|
body = response.json()
|
|
assert body["owner"]
|
|
assert body["repo_name"]
|
|
assert body["clone_url"]
|
|
# A tenant repository must never be created public.
|
|
assert body.get("private", True) is True
|
|
|
|
|
|
async def test_repo_status_is_reported_after_ensure(client: AsyncClient):
|
|
token = (await signup(client, "repo-status@example.com"))["access_token"]
|
|
org_id, project_id = await _org_and_project(client, token)
|
|
base = f"/api/organizations/{org_id}/projects/{project_id}"
|
|
|
|
# Before ensure, the platform has no record of a repository.
|
|
before = await client.get(f"{base}/repo/status", headers=auth_header(token))
|
|
assert before.status_code == 200, before.text
|
|
assert before.json()["exists"] is False
|
|
|
|
await client.post(f"{base}/repo/ensure", headers=auth_header(token))
|
|
|
|
after = await client.get(f"{base}/repo/status", headers=auth_header(token))
|
|
assert after.status_code == 200, after.text
|
|
assert after.json()["exists"] is True
|
|
|
|
|
|
async def test_git_password_never_leaves_the_api(client: AsyncClient):
|
|
"""`ensure_credentials` returns plaintext to the service layer only.
|
|
|
|
The regression this guards against is a future refactor adding the
|
|
credential object straight to a response model.
|
|
"""
|
|
token = (await signup(client, "repo-secret@example.com"))["access_token"]
|
|
org_id, project_id = await _org_and_project(client, token)
|
|
base = f"/api/organizations/{org_id}/projects/{project_id}"
|
|
|
|
await client.post(f"{base}/repo/ensure", headers=auth_header(token))
|
|
|
|
for path in (f"{base}/repo", f"{base}/repo/status"):
|
|
response = await client.get(path, headers=auth_header(token))
|
|
assert response.status_code == 200, response.text
|
|
payload = json.dumps(response.json())
|
|
for forbidden in ("password", "encrypted_password", "token"):
|
|
assert forbidden not in payload.lower(), f"{forbidden} leaked via {path}"
|
|
|
|
|
|
async def test_repository_of_another_tenant_is_not_reachable(client: AsyncClient):
|
|
"""Tenant isolation must hold on the repo sub-resources too."""
|
|
token_a = (await signup(client, "repo-a@example.com"))["access_token"]
|
|
token_b = (await signup(client, "repo-b@example.com"))["access_token"]
|
|
org_a, project_a = await _org_and_project(client, token_a)
|
|
|
|
# B is authenticated but not a member of A's organization.
|
|
response = await client.post(
|
|
f"/api/organizations/{org_a}/projects/{project_a}/repo/ensure",
|
|
headers=auth_header(token_b),
|
|
)
|
|
# 404 rather than 403, so the API does not confirm the project exists.
|
|
assert response.status_code == 404, response.text
|
|
|
|
|
|
async def test_ensure_repository_is_idempotent(client: AsyncClient):
|
|
token = (await signup(client, "repo-idem@example.com"))["access_token"]
|
|
org_id, project_id = await _org_and_project(client, token)
|
|
base = f"/api/organizations/{org_id}/projects/{project_id}"
|
|
|
|
first = await client.post(f"{base}/repo/ensure", headers=auth_header(token))
|
|
second = await client.post(f"{base}/repo/ensure", headers=auth_header(token))
|
|
assert first.status_code in (200, 201), first.text
|
|
assert second.status_code in (200, 201), second.text
|
|
assert first.json()["repo_name"] == second.json()["repo_name"]
|