"""Mock adapters for the non-AI ports, so the stack runs with zero credentials. These are not toys: they enforce the same contracts (timeouts, output caps, validation) as the real adapters, so switching a ``*_PROVIDER`` env var does not change application behaviour. The model port has no mock — ``agy`` is the only model provider (see ``adapters/agy.py``). """ from __future__ import annotations import hashlib import secrets import time import uuid from collections.abc import Iterable from typing import Any from app.application.ports import ( BranchInfo, CheckoutSession, CommandResult, CommitInfo, GitCredentials, MergeResult, Notification, PaymentResult, PreviewDomainInfo, PullRequestInfo, RepoInfo, RepoStatus, ReverseProxyResult, SessionInfo, SnapshotInfo, StoredObject, VectorHit, WorkspaceInfo, ) from app.domain.platform.rules import ( assert_port_allocatable, build_host, validate_managed_host, validate_proxy_target, ) # --------------------------------------------------------------------------- # Git # --------------------------------------------------------------------------- class MockGitAdapter: """In-memory Git provider. Nothing touches a real forge. ``owner`` / ``repo_name`` are accepted for interface parity with the Gitea adapter but ignored here — this adapter keys everything by ``project_id``. """ def __init__(self) -> None: self._repos: dict[str, RepoInfo] = {} self._credentials: dict[str, GitCredentials] = {} self._branches: dict[str, dict[str, str]] = {} self._prs: dict[str, list[PullRequestInfo]] = {} async def ensure_project_repo( self, *, project_id: str, repo_name: str, description: str = "" ) -> RepoInfo: if project_id not in self._repos: self._repos[project_id] = RepoInfo( owner="mock", repo_name=repo_name, clone_url=f"https://git.example.invalid/mock/{repo_name}.git", web_url=f"https://git.example.invalid/mock/{repo_name}", default_branch="main", private=True, ) self._branches[project_id] = {"main": hashlib.sha1(b"init").hexdigest()[:40]} self._prs[project_id] = [] return self._repos[project_id] async def ensure_credentials( self, *, project_id: str, owner: str = "", repo_name: str = "" ) -> GitCredentials: if project_id not in self._credentials: self._credentials[project_id] = GitCredentials( username=f"proj_{project_id[:8]}", password=secrets.token_urlsafe(12), ) return self._credentials[project_id] async def create_branch( self, *, project_id: str, branch_name: str, from_branch: str, owner: str = "", repo_name: str = "", ) -> BranchInfo: branches = self._branches.setdefault(project_id, {"main": "0" * 40}) base = branches.get(from_branch, "0" * 40) sha = hashlib.sha1(f"{base}:{branch_name}".encode()).hexdigest()[:40] branches[branch_name] = sha return BranchInfo(name=branch_name, sha=sha) async def commit_files( self, *, project_id: str, branch_name: str, files: dict[str, str], message: str, owner: str = "", repo_name: str = "", ) -> CommitInfo: digest = hashlib.sha1( f"{branch_name}:{message}:{sorted(files)}".encode() ).hexdigest()[:40] self._branches.setdefault(project_id, {})[branch_name] = digest return CommitInfo( sha=digest, branch=branch_name, message=message, files_changed=len(files) ) async def open_pull_request( self, *, project_id: str, head: str, base: str, title: str, body: str = "", owner: str = "", repo_name: str = "", ) -> PullRequestInfo: prs = self._prs.setdefault(project_id, []) number = len(prs) + 1 repo = self._repos.get(project_id) web = repo.web_url if repo else "https://git.example.invalid/mock" pr = PullRequestInfo(number=number, url=f"{web}/pulls/{number}", state="open") prs.append(pr) return pr async def merge_pull_request( self, *, project_id: str, pr_number: int, owner: str = "", repo_name: str = "" ) -> MergeResult: prs = self._prs.setdefault(project_id, []) for pr in prs: if pr.number == pr_number: pr.state = "merged" return MergeResult(merged=True, message="ادغام شد (mock).") return MergeResult(merged=False, message="PR یافت نشد.") async def get_repo_status( self, *, project_id: str, owner: str = "", repo_name: str = "" ) -> RepoStatus: repo = self._repos.get(project_id) if repo is None: return RepoStatus(exists=False) return RepoStatus( exists=True, private=repo.private, default_branch=repo.default_branch, branches=sorted(self._branches.get(project_id, {}).keys()), open_pull_requests=sum( 1 for pr in self._prs.get(project_id, []) if pr.state == "open" ), ) # --------------------------------------------------------------------------- # Domain router # --------------------------------------------------------------------------- class MockDomainRouterAdapter: """In-memory port + subdomain ledger. Applies the *same* validation rules as the real Caddy adapter, so tests that pass here describe behaviour the production adapter also guarantees. """ def __init__( self, *, root_domain: str = "msa.artacloud.ir", port_start: int = 8080, port_end: int = 8999, ) -> None: self.root_domain = root_domain self.port_start = port_start self.port_end = port_end self._ports: dict[tuple[str, str], int] = {} self._domains: dict[str, PreviewDomainInfo] = {} self._project_domains: dict[str, list[str]] = {} async def allocate_port( self, *, organization_id: str, project_id: str, service_name: str ) -> int: key = (project_id, service_name) if key in self._ports: return self._ports[key] used = set(self._ports.values()) for port in range(self.port_start, self.port_end + 1): if port in used: continue try: assert_port_allocatable(port, start=self.port_start, end=self.port_end) except Exception: continue self._ports[key] = port return port raise RuntimeError("No available ports in the configured range.") async def claim_preview_domain( self, *, organization_id: str, project_id: str, requested_subdomain: str, port: int, ) -> PreviewDomainInfo: host = build_host(requested_subdomain, self.root_domain) if host in self._domains: owner_domains = self._project_domains.get(project_id, []) if host not in owner_domains: raise ValueError(f"دامنهٔ {host} قبلاً ثبت شده است.") info = PreviewDomainInfo( host=host, port=port, url=f"https://{host}", status="active" ) self._domains[host] = info self._project_domains.setdefault(project_id, []).append(host) return info async def configure_reverse_proxy(self, *, host: str, port: int) -> ReverseProxyResult: validate_managed_host(host, self.root_domain) validate_proxy_target(f"localhost:{port}") return ReverseProxyResult(ok=True, message="route ثبت شد (mock).", url=f"https://{host}") async def remove_preview_domain(self, *, domain_id: str) -> None: for host, info in list(self._domains.items()): if info.host == domain_id or domain_id in self._project_domains.get(domain_id, []): self._domains.pop(host, None) self._domains.pop(domain_id, None) for hosts in self._project_domains.values(): if domain_id in hosts: hosts.remove(domain_id) async def list_project_domains(self, *, project_id: str) -> list[PreviewDomainInfo]: return [ self._domains[h] for h in self._project_domains.get(project_id, []) if h in self._domains ] # --------------------------------------------------------------------------- # Sandbox # --------------------------------------------------------------------------- class LocalMockSandboxAdapter: """Executes nothing. Returns plausible results and records commands. ``run_command`` still honours the timeout and output cap contract so callers cannot depend on behaviour the real adapters will not provide. """ def __init__(self, max_output_bytes: int = 200_000) -> None: self._workspaces: dict[str, str] = {} self._sessions: dict[str, str] = {} self._files: dict[str, dict[str, str]] = {} self.commands: list[str] = [] self.max_output_bytes = max_output_bytes async def create_workspace( self, *, organization_id: str, project_id: str, template_id: str, image: str ) -> WorkspaceInfo: external_id = f"ws_{uuid.uuid4().hex[:12]}" self._workspaces[external_id] = image return WorkspaceInfo(external_id=external_id, status="ready", raw={"image": image}) async def start_session(self, *, workspace_external_id: str) -> SessionInfo: external_id = f"ses_{uuid.uuid4().hex[:12]}" self._sessions[external_id] = workspace_external_id self._files.setdefault(external_id, {}) return SessionInfo(external_id=external_id, status="running") async def run_command( self, *, session_external_id: str, command: str, timeout_seconds: int, max_output_bytes: int | None = None, env: dict[str, str] | None = None, ) -> CommandResult: limit = max_output_bytes or self.max_output_bytes started = time.perf_counter() self.commands.append(command) stdout = f"[mock] اجرا شد: {command}" truncated = len(stdout.encode()) > limit if truncated: stdout = stdout.encode()[:limit].decode(errors="ignore") return CommandResult( exit_code=0, stdout=stdout, stderr="", duration_ms=max(1, int((time.perf_counter() - started) * 1000)), truncated=truncated, ) async def read_file(self, *, session_external_id: str, path: str) -> str: return self._files.get(session_external_id, {}).get(path, "") async def write_file(self, *, session_external_id: str, path: str, content: str) -> None: self._files.setdefault(session_external_id, {})[path] = content async def snapshot(self, *, session_external_id: str, name: str) -> SnapshotInfo: return SnapshotInfo(external_id=f"snap_{uuid.uuid4().hex[:12]}", name=name) async def pause(self, *, session_external_id: str) -> None: return None async def resume(self, *, session_external_id: str) -> SessionInfo: return SessionInfo(external_id=session_external_id, status="running") async def destroy(self, *, session_external_id: str) -> None: self._sessions.pop(session_external_id, None) self._files.pop(session_external_id, None) async def get_preview_url(self, *, session_external_id: str, port: int) -> str: return f"http://127.0.0.1:{port}" # --------------------------------------------------------------------------- # Payment # --------------------------------------------------------------------------- class LocalMockPaymentAdapter: async def create_checkout( self, *, organization_id: str, amount: float, currency: str, success_url: str, cancel_url: str, ) -> CheckoutSession: external_id = f"mock_cs_{uuid.uuid4().hex[:12]}" return CheckoutSession( external_id=external_id, url=f"{success_url}?mock_checkout={external_id}", amount=amount, currency=currency, ) async def refund( self, *, external_payment_id: str, amount: float | None = None ) -> PaymentResult: return PaymentResult(ok=True, external_id=external_payment_id, message="بازگشت (mock).") # --------------------------------------------------------------------------- # Storage / vector / notification # --------------------------------------------------------------------------- class InMemoryStorageAdapter: def __init__(self) -> None: self._objects: dict[str, bytes] = {} async def put(self, *, key: str, data: bytes, content_type: str = "") -> StoredObject: self._objects[key] = data return StoredObject(key=key, size_bytes=len(data), url=f"memory://{key}") async def get(self, *, key: str) -> bytes: return self._objects.get(key, b"") async def delete(self, *, key: str) -> None: self._objects.pop(key, None) async def signed_url(self, *, key: str, expires_seconds: int = 900) -> str: return f"memory://{key}?expires={expires_seconds}" class InMemoryVectorStoreAdapter: """Keyword-overlap search. Good enough for MVP; swappable for pgvector.""" def __init__(self) -> None: self._chunks: dict[str, tuple[str, str, str, dict]] = {} async def upsert( self, *, organization_id: str, chunks: Iterable[tuple[str, str, list[float]]], metadata: dict | None = None, ) -> int: count = 0 for chunk_id, content, _embedding in chunks: document_id = str((metadata or {}).get("document_id", "")) self._chunks[chunk_id] = (organization_id, document_id, content, metadata or {}) count += 1 return count async def search( self, *, organization_id: str, query: str, limit: int = 5, project_id: str | None = None, ) -> list[VectorHit]: terms = {t for t in query.lower().split() if t} scored: list[VectorHit] = [] for chunk_id, (org, document_id, content, meta) in self._chunks.items(): if org != organization_id: continue if project_id and meta.get("project_id") not in (None, project_id): continue overlap = len(terms & set(content.lower().split())) if overlap: scored.append( VectorHit( chunk_id=chunk_id, document_id=document_id, content=content, score=float(overlap), metadata=meta, ) ) scored.sort(key=lambda h: h.score, reverse=True) return scored[:limit] async def delete_document(self, *, document_id: str) -> None: for chunk_id, (_, doc_id, _, _) in list(self._chunks.items()): if doc_id == document_id: self._chunks.pop(chunk_id, None) class ConsoleNotificationAdapter: """Records notifications instead of sending them.""" def __init__(self) -> None: self.sent: list[Notification] = [] async def send(self, *, notification: Notification) -> bool: self.sent.append(notification) return True class InMemoryAuditAdapter: """In-memory AuditPort, used by unit tests that do not want a database.""" def __init__(self) -> None: # Records whatever the caller hands over; the port is untyped here. self.events: list[Any] = [] async def record(self, *, event): self.events.append(event) return event async def list_for_organization( self, *, organization_id: str, limit: int = 100, offset: int = 0 ): matching = [e for e in self.events if e.organization_id == organization_id] return matching[offset : offset + limit]