mirror of
https://github.com/kuhyx/testsAndMisc.git
synced 2026-07-04 13:03:13 +02:00
Krzysztof kuhy Rudnicki
72c6c3788c
- Move puzzle_solver/, poker_modifier_app/, articles/, tests/ into python_pkg/ - Move moviepy_showcase.py and _moviepy_*.py into python_pkg/moviepy_showcase/ - Update all imports to use python_pkg. prefix - Update pyproject.toml per-file-ignores and pytest testpaths - Add pre-commit hook to enforce Python files under python_pkg/
110 lines
3.3 KiB
Python
110 lines
3.3 KiB
Python
"""Integration tests for the articles C server API."""
|
|
|
|
from http import HTTPStatus
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import socket
|
|
import subprocess
|
|
import time
|
|
from typing import Any
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
import pytest
|
|
|
|
|
|
def _req(
|
|
url: str, method: str = "GET", data: dict[str, Any] | bytes | None = None
|
|
) -> tuple[int, bytes]:
|
|
"""Send an HTTP request and return status code and body."""
|
|
if data is not None and not isinstance(data, bytes | bytearray):
|
|
data = json.dumps(data).encode("utf-8")
|
|
req = urllib.request.Request(url, data=data, method=method)
|
|
req.add_header("Content-Type", "application/json")
|
|
with urllib.request.urlopen(req, timeout=5) as resp:
|
|
body = resp.read()
|
|
return resp.getcode(), body
|
|
|
|
|
|
def test_crud_roundtrip(tmp_path: Path) -> None:
|
|
"""Test full CRUD lifecycle for articles API."""
|
|
# Build C server
|
|
here = Path(__file__).resolve().parent
|
|
subprocess.run(["make", "-s", "server_c"], check=True, cwd=str(here))
|
|
|
|
# Find a free port
|
|
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
|
|
s.bind(("127.0.0.1", 0))
|
|
_, port = s.getsockname()
|
|
host = "127.0.0.1"
|
|
base = f"http://{host}:{port}"
|
|
|
|
# Isolate storage and start server
|
|
env = os.environ.copy()
|
|
env["ARTICLES_DATA_DIR"] = str(tmp_path)
|
|
env["HOST"] = host
|
|
env["PORT"] = str(port)
|
|
srv = subprocess.Popen(["./server_c"], cwd=str(here), env=env)
|
|
try:
|
|
# wait briefly for server to be ready
|
|
for _ in range(30):
|
|
try:
|
|
with urllib.request.urlopen(
|
|
base + "/api/articles", timeout=0.2
|
|
) as resp:
|
|
resp.read()
|
|
break
|
|
except (OSError, urllib.error.URLError):
|
|
time.sleep(0.05)
|
|
|
|
# Create
|
|
code, body = _req(
|
|
base + "/api/articles",
|
|
method="POST",
|
|
data={
|
|
"title": "T1",
|
|
"body": "<p>Hello</p>",
|
|
"thumb": "data:image/png;base64,xyz",
|
|
},
|
|
)
|
|
assert code == HTTPStatus.CREATED
|
|
created = json.loads(body)
|
|
art_id = created["id"]
|
|
|
|
# List
|
|
code, body = _req(base + "/api/articles")
|
|
assert code == HTTPStatus.OK
|
|
items = json.loads(body)
|
|
assert any(a["id"] == art_id for a in items)
|
|
|
|
# Get one
|
|
code, body = _req(base + f"/api/articles/{art_id}")
|
|
assert code == HTTPStatus.OK
|
|
got = json.loads(body)
|
|
assert got["title"] == "T1"
|
|
|
|
# Update
|
|
code, body = _req(
|
|
base + f"/api/articles/{art_id}", method="PUT", data={"title": "T2"}
|
|
)
|
|
assert code == HTTPStatus.OK
|
|
updated = json.loads(body)
|
|
assert updated["title"] == "T2"
|
|
|
|
# Delete
|
|
code, _ = _req(base + f"/api/articles/{art_id}", method="DELETE")
|
|
assert code == HTTPStatus.NO_CONTENT
|
|
|
|
# Ensure gone
|
|
with pytest.raises(urllib.error.HTTPError) as exc_info:
|
|
_req(base + f"/api/articles/{art_id}")
|
|
assert exc_info.value.code == HTTPStatus.NOT_FOUND
|
|
|
|
finally:
|
|
srv.terminate()
|
|
try:
|
|
srv.wait(timeout=2)
|
|
except subprocess.TimeoutExpired:
|
|
srv.kill()
|