155 lines
5.2 KiB
Python
155 lines
5.2 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import socket
|
|
import subprocess
|
|
import time
|
|
import unittest
|
|
from pathlib import Path
|
|
from urllib.error import HTTPError, URLError
|
|
from urllib.request import Request, urlopen
|
|
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
|
API_WORKDIR = PROJECT_ROOT / "journal-master" / "journal"
|
|
API_PROJECT = API_WORKDIR / "Journal.Api" / "Journal.Api.csproj"
|
|
API_DLL = API_WORKDIR / "Journal.Api" / "bin" / "Debug" / "net10.0" / "Journal.Api.dll"
|
|
|
|
|
|
def _pick_free_port() -> int:
|
|
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
|
|
sock.bind(("127.0.0.1", 0))
|
|
return int(sock.getsockname()[1])
|
|
|
|
|
|
def _http_json(method: str, url: str, body: str | None = None) -> tuple[int, dict]:
|
|
headers = {"Accept": "application/json"}
|
|
data = None
|
|
if body is not None:
|
|
headers["Content-Type"] = "application/json"
|
|
data = body.encode("utf-8")
|
|
|
|
request = Request(url=url, method=method, data=data, headers=headers)
|
|
try:
|
|
with urlopen(request, timeout=5.0) as response:
|
|
payload = response.read().decode("utf-8")
|
|
return int(response.status), json.loads(payload)
|
|
except HTTPError as ex:
|
|
payload = ex.read().decode("utf-8")
|
|
try:
|
|
parsed = json.loads(payload)
|
|
except json.JSONDecodeError:
|
|
parsed = {"raw": payload}
|
|
return int(ex.code), parsed
|
|
|
|
|
|
class ApiContractTests(unittest.TestCase):
|
|
process: subprocess.Popen[str] | None = None
|
|
base_url: str = ""
|
|
|
|
@classmethod
|
|
def setUpClass(cls) -> None:
|
|
if not API_DLL.exists():
|
|
raise FileNotFoundError(
|
|
f"Missing API binary: {API_DLL}. Build Journal.Api first."
|
|
)
|
|
|
|
port = _pick_free_port()
|
|
cls.base_url = f"http://127.0.0.1:{port}"
|
|
env = os.environ.copy()
|
|
env["DOTNET_CLI_HOME"] = str(PROJECT_ROOT / ".dotnet_home")
|
|
env["NUGET_PACKAGES"] = str(PROJECT_ROOT / ".nuget" / "packages")
|
|
env["NUGET_HTTP_CACHE_PATH"] = str(PROJECT_ROOT / ".nuget" / "http-cache")
|
|
env["DOTNET_SKIP_FIRST_TIME_EXPERIENCE"] = "1"
|
|
env["DOTNET_GENERATE_ASPNET_CERTIFICATE"] = "0"
|
|
env["DOTNET_CLI_TELEMETRY_OPTOUT"] = "1"
|
|
env["ASPNETCORE_URLS"] = cls.base_url
|
|
|
|
cls.process = subprocess.Popen(
|
|
[
|
|
"dotnet",
|
|
str(API_DLL),
|
|
],
|
|
cwd=str(API_WORKDIR),
|
|
env=env,
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
text=True,
|
|
)
|
|
|
|
deadline = time.time() + 60
|
|
while time.time() < deadline:
|
|
if cls.process.poll() is not None:
|
|
raise RuntimeError("Journal.Api exited during startup.")
|
|
try:
|
|
status, payload = _http_json("GET", f"{cls.base_url}/health")
|
|
if status == 200 and payload.get("ok") is True:
|
|
return
|
|
except (URLError, TimeoutError):
|
|
pass
|
|
time.sleep(0.5)
|
|
|
|
raise TimeoutError("Timed out waiting for Journal.Api /health endpoint.")
|
|
|
|
@classmethod
|
|
def tearDownClass(cls) -> None:
|
|
if cls.process is None:
|
|
return
|
|
if cls.process.poll() is None:
|
|
cls.process.terminate()
|
|
try:
|
|
cls.process.wait(timeout=10)
|
|
except subprocess.TimeoutExpired:
|
|
cls.process.kill()
|
|
cls.process = None
|
|
|
|
def test_health_endpoint_envelope(self):
|
|
status, payload = _http_json("GET", f"{self.base_url}/health")
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual(payload, {"ok": True, "data": "healthy"})
|
|
|
|
def test_command_success_path(self):
|
|
status, payload = _http_json(
|
|
"POST",
|
|
f"{self.base_url}/api/command",
|
|
body=json.dumps({"action": "config.get", "payload": {}}),
|
|
)
|
|
self.assertEqual(status, 200)
|
|
self.assertTrue(payload.get("ok"))
|
|
self.assertIsInstance(payload.get("data"), dict)
|
|
|
|
def test_command_unknown_action_error_envelope(self):
|
|
status, payload = _http_json(
|
|
"POST",
|
|
f"{self.base_url}/api/command",
|
|
body=json.dumps({"action": "unknown.action", "payload": {}}),
|
|
)
|
|
self.assertEqual(status, 200)
|
|
self.assertFalse(payload.get("ok"))
|
|
self.assertIn("unknown action", str(payload.get("error", "")).lower())
|
|
|
|
def test_command_missing_payload_error_envelope(self):
|
|
status, payload = _http_json(
|
|
"POST",
|
|
f"{self.base_url}/api/command",
|
|
body=json.dumps({"action": "entries.save"}),
|
|
)
|
|
self.assertEqual(status, 200)
|
|
self.assertFalse(payload.get("ok"))
|
|
self.assertIn("payload", str(payload.get("error", "")).lower())
|
|
|
|
def test_command_malformed_json_error_envelope(self):
|
|
status, payload = _http_json(
|
|
"POST",
|
|
f"{self.base_url}/api/command",
|
|
body='{"action":',
|
|
)
|
|
self.assertEqual(status, 200)
|
|
self.assertFalse(payload.get("ok"))
|
|
self.assertIn("invalid command json", str(payload.get("error", "")).lower())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|