diff --git a/.github/workflows/integration-functional.yml b/.github/workflows/integration-functional.yml new file mode 100644 index 0000000..75be2fe --- /dev/null +++ b/.github/workflows/integration-functional.yml @@ -0,0 +1,45 @@ +# SPDX-FileCopyrightText: 2026 Andrew Zhang +# +# SPDX-License-Identifier: BSL-1.0 + +name: Integration functional + +on: + push: + branches: [main, develop] + pull_request: + branches: [main, develop] + +jobs: + integration-functional: + runs-on: ubuntu-latest + timeout-minutes: 25 + steps: + # actions/checkout v6.0.2 + - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd + with: + persist-credentials: false + + # actions/setup-python v6.2.0 + - uses: actions/setup-python@a309ff8b426b58ec0e2a45f0f869d46889d02405 + with: + python-version: '3.12' + + # astral-sh/setup-uv v8.1.0 + - uses: astral-sh/setup-uv@08807647e7069bb48b6ef5acd8ec9567f424441b + with: + version: 0.11.12 + + - name: Run integration functional tests + env: + # Use a repository *secret* (Settings → Secrets → Actions), not a variable. + GH_TEST_REPO_TOKEN: ${{ secrets.GH_TEST_REPO_TOKEN }} + run: bash scripts/integration-functional.sh + + - name: Upload logs on failure + if: failure() + # actions/upload-artifact v4.6.2 + uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 + with: + name: integration-functional-logs + path: /tmp/compose-logs.txt diff --git a/.github/workflows/integration-smoke.yml b/.github/workflows/integration-smoke.yml index c2860a9..5112f6c 100644 --- a/.github/workflows/integration-smoke.yml +++ b/.github/workflows/integration-smoke.yml @@ -25,6 +25,11 @@ jobs: with: python-version: '3.12' + # astral-sh/setup-uv v8.1.0 + - uses: astral-sh/setup-uv@08807647e7069bb48b6ef5acd8ec9567f424441b + with: + version: 0.11.12 + - name: Run integration smoke tests run: bash scripts/integration-smoke.sh diff --git a/pyproject.toml b/pyproject.toml index 943c67e..4d0507c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -11,6 +11,10 @@ dev = [ "coverage[toml]>=7.6.0", "pytest-cov>=7.1.0" ] +integration = [ + "pytest>=8.3", + "pytest-timeout>=2.3.1" +] lint = [ {include-group = "pre-commit"} ] diff --git a/scripts/integration-functional.sh b/scripts/integration-functional.sh new file mode 100755 index 0000000..2422b7e --- /dev/null +++ b/scripts/integration-functional.sh @@ -0,0 +1,61 @@ +#!/usr/bin/env bash +# SPDX-FileCopyrightText: 2026 Andrew Zhang +# SPDX-License-Identifier: BSL-1.0 + +# Integration functional test entrypoint (P1). +# Builds the stack, waits for health, creates API token, extracts SSH pubkey, +# runs functional tests against a live Weblate instance. + +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" +# shellcheck source=lib/weblate-stack.sh +source "${SCRIPT_DIR}/lib/weblate-stack.sh" + +cleanup() { + local exit_code=$? + set +e + echo "--- Collecting logs ---" + stack_logs /tmp/compose-logs.txt + echo "--- Tearing down stack ---" + stack_down + exit "$exit_code" +} +trap cleanup EXIT + +echo "=== Building stack ===" +stack_build + +echo "=== Starting stack ===" +stack_up + +echo "=== Waiting for Weblate ===" +stack_wait_healthy "${HEALTH_TIMEOUT:-180}" + +echo "=== Creating API token ===" +WEBLATE_API_TOKEN="$(stack_create_token admin)" +export WEBLATE_API_TOKEN +export WEBLATE_LIVE_BASE_URL="${WEBLATE_LIVE_BASE_URL:-http://localhost:${WEBLATE_PORT:-8080}}" +export WEBLATE_COMPOSE_FILE="${COMPOSE_FILE}" +export WEBLATE_COMPOSE_PROJECT="${COMPOSE_PROJECT_NAME}" + +echo "=== Extracting Weblate SSH public key ===" +TMP_WEBLATE_SSH_PUBKEY="$(compose exec -T weblate cat /app/data/ssh/id_rsa.pub)" +if [[ -z "${TMP_WEBLATE_SSH_PUBKEY}" ]]; then + echo "ERROR: Failed to read Weblate SSH public key from container." >&2 + exit 1 +fi +export WEBLATE_SSH_PUBKEY="${TMP_WEBLATE_SSH_PUBKEY}" +unset TMP_WEBLATE_SSH_PUBKEY + +if [[ -n "${GH_TEST_REPO_TOKEN:-}" ]]; then + export GH_TEST_REPO_TOKEN + echo "=== GH_TEST_REPO_TOKEN is set (${#GH_TEST_REPO_TOKEN} chars); GitHub E2E tests enabled ===" +else + echo "=== GH_TEST_REPO_TOKEN is not set; GitHub E2E/Celery tests will be skipped ===" +fi + +echo "=== Running functional tests ===" +uv pip install --quiet --system --group integration +python -m pytest --confcutdir=tests/integration --override-ini addopts= \ + tests/integration/test_functional.py -v --timeout=300 diff --git a/scripts/integration-smoke.sh b/scripts/integration-smoke.sh index d809e49..52f70c4 100755 --- a/scripts/integration-smoke.sh +++ b/scripts/integration-smoke.sh @@ -40,6 +40,8 @@ export WEBLATE_COMPOSE_FILE="${COMPOSE_FILE}" export WEBLATE_COMPOSE_PROJECT="${COMPOSE_PROJECT_NAME}" echo "=== Running smoke tests ===" -pip install --quiet pytest +# Same pytest floor as dev/pre-commit (pyproject.toml); not pinned to uv.lock. +# --system: setup-python on CI has no project venv (matches ci-dependencies.yml). +uv pip install --quiet --system --group integration # Do not load tests/conftest.py (Django host setup); integration tests only need pytest + stdlib. python -m pytest --confcutdir=tests/integration --override-ini addopts= tests/integration/test_smoke.py -v diff --git a/tests/endpoint/test_serializers.py b/tests/endpoint/test_serializers.py index ddb3732..8d0d8d1 100644 --- a/tests/endpoint/test_serializers.py +++ b/tests/endpoint/test_serializers.py @@ -27,7 +27,7 @@ def test_add_or_update_serializer_accepts_extensions() -> None: data={ "organization": "o", "version": "v", - "add_or_update": {"ja": ["unordered"]}, + "add_or_update": {"zh_Hans": ["unordered"]}, "extensions": [".adoc", ".md"], } ) @@ -64,11 +64,11 @@ def test_add_or_update_serializer_rejects_non_list_submodules() -> None: data={ "organization": "o", "version": "v", - "add_or_update": {"ja": "json"}, + "add_or_update": {"zh_Hans": "json"}, } ) assert not ser.is_valid() - assert "ja" in ser.errors["add_or_update"] + assert "zh_Hans" in ser.errors["add_or_update"] def test_add_or_update_serializer_missing_required_fields() -> None: diff --git a/tests/endpoint/test_views.py b/tests/endpoint/test_views.py index 612358e..7bccf21 100644 --- a/tests/endpoint/test_views.py +++ b/tests/endpoint/test_views.py @@ -71,7 +71,7 @@ def test_add_or_update_requires_authentication( { "organization": "o", "version": "v", - "add_or_update": {"ja": ["json"]}, + "add_or_update": {"zh_Hans": ["json"]}, }, format="json", ) @@ -98,7 +98,7 @@ def test_add_or_update_accepts_and_enqueues_like_boost_weblate( { "organization": "o", "version": "v", - "add_or_update": {"ja": ["json"]}, + "add_or_update": {"zh_Hans": ["json"]}, }, format="json", ) @@ -122,7 +122,7 @@ def test_add_or_update_accepts_and_enqueues_like_boost_weblate( delay_mock.assert_called_once_with( organization="o", - add_or_update={"ja": ["json"]}, + add_or_update={"zh_Hans": ["json"]}, version="v", extensions=None, user_id=42, @@ -168,16 +168,16 @@ def process_all(self, submodules, *, user, request=None): # noqa: ANN001 result = tasks_mod.boost_add_or_update_task.run( organization="org", - add_or_update={"ja": ["json"], "zh": ["a"]}, + add_or_update={"zh_Hans": ["a"], "ja": ["json"]}, version="boost-1.0", extensions=[".md"], user_id=7, ) get_mock.assert_called_once_with(pk=7) - assert calls == [("ja", ["json"]), ("zh", ["a"])] + assert calls == [("zh_Hans", ["a"]), ("ja", ["json"])] + assert result["zh_Hans"]["organization"] == "org" assert result["ja"]["submodules"] == ["json"] - assert result["zh"]["organization"] == "org" def test_boost_add_or_update_task_propagates_service_errors( diff --git a/tests/fixtures/asciidoc_fixture.adoc b/tests/fixtures/asciidoc_fixture.adoc new file mode 100644 index 0000000..4ad7873 --- /dev/null +++ b/tests/fixtures/asciidoc_fixture.adoc @@ -0,0 +1,88 @@ +// Synthetic fixture for AsciiDoc parser tests. +// Patterns adapted from Beast documentation (sections, +// headings, tables with source blocks, description lists, admonitions, links). + += Complex AsciiDoc test fixture + +[#complex_fixture] +== Complex AsciiDoc test fixture + +This opening paragraph soft-wraps onto a second line while keeping an +https://example.com/path[RFC-style link] and a named +xref:message[`message`] reference in the same paragraph block. + +== Section headings and lists + +* First bullet names xref:request[`request`]. +* Second bullet continues the list with plain prose. + +[[anchor_skipped]] + +[#custom] +=== Custom heading with id + +[quote] +____ +This is a single-line blockquote for translation. +____ + +.NOTE +==== +This is a multi-line admonition body. It contains a paragraph that should +still be extracted when the note block is parsed recursively. + +A second paragraph inside the same note uses +https://tools.ietf.org/html/rfc6455[WebSocket] markup. +==== + +[#nested_inner] +== Nested section title here + +Inner section prose explains that `template` parameters accept any +xref:fields[`fields`] type meeting requirements. +The paragraph joins wrapped source lines into one translatable unit. + +[source] +---- +// This indented line starts a code block (non-translatable). +// It must not become its own translation unit. +---- + +After the code block, prose resumes with an image that is skipped: + +image::beast/images/message.png[width=100px,height=50px] + +.Message patterns +|=== +|Name |Description + +|__message__ +a|``` +/// Class template overview +template +class message; +``` + +|xref:request[`request`] +a|``` +/// HTTP request alias +template +using request = message; +``` + +|Plain prose cell +|This cell has human-readable text only, without a source block. +|=== + +[qanda] +Does this fixture include a description list?:: +Yes. This pair mimics patterns from the FAQ chapter: quoted question +strings and multi-sentence answers with blank-line-separated paragraphs. + +Second paragraph in the same answer cell. + +What about table titles?:: +The table above includes an explicit title line before the opening delimiter. ++ + +.WARNING: This is a one-line warning about edge cases in handshakes. diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index 3d09914..4401ee3 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -2,55 +2,75 @@ # # SPDX-License-Identifier: BSL-1.0 -"""Shared fixtures for integration tests.""" +"""Shared fixtures for integration tests (smoke + functional).""" from __future__ import annotations import os from collections.abc import Callable -from typing import Any +from pathlib import Path import pytest -from tests.integration.lib.docker_exec import ( - docker_exec_python, - docker_exec_python_json, -) -from tests.integration.lib.http import base_url as _base_url -from tests.integration.lib.http import http_get +from tests.integration.lib.docker_exec import docker_exec_python, docker_exec_read_file +from tests.integration.lib.gh_repo import EphemeralGitHubRepo, default_repo_name +from tests.integration.lib.http import base_url +from tests.integration.lib.weblate_api import WeblateAPI + +FIXTURES_DIR = Path(__file__).resolve().parent.parent / "fixtures" +TEST_LANG_CODE = "zh_Hans" +TEST_BRANCH = f"local-{TEST_LANG_CODE}" +TEST_VERSION = "test-1.0.0" + + +@pytest.fixture(scope="session") +def live_base_url() -> str: + return base_url() @pytest.fixture(scope="session") def api_token() -> str: - token = os.environ.get("WEBLATE_API_TOKEN") + token = os.environ.get("WEBLATE_API_TOKEN", "").strip() if not token: - pytest.skip("WEBLATE_API_TOKEN not set") + pytest.skip("WEBLATE_API_TOKEN is not set") return token @pytest.fixture(scope="session") -def live_base_url() -> str: - return _base_url() +def exec_python() -> Callable[[str], str]: + return docker_exec_python @pytest.fixture(scope="session") -def authed_get(api_token: str) -> Callable[..., tuple[int, Any]]: # noqa: E501 - """GET helper pre-bound with the API token.""" - token = api_token - - def _get(path: str, **kwargs: Any) -> tuple[int, Any]: - return http_get(path, token=token, **kwargs) - - return _get +def weblate_api(api_token: str, live_base_url: str) -> WeblateAPI: + return WeblateAPI(api_token, live_base_url=live_base_url) @pytest.fixture(scope="session") -def exec_python() -> Callable[[str], str]: - """Execute a Python snippet inside the Weblate container.""" - return docker_exec_python +def weblate_ssh_pubkey() -> str: + pubkey = os.environ.get("WEBLATE_SSH_PUBKEY", "").strip() + if pubkey: + return pubkey + return docker_exec_read_file("/app/data/ssh/id_rsa.pub") @pytest.fixture(scope="session") -def exec_python_json() -> Callable[[str], object]: - """Execute a Python snippet inside the container and parse JSON output.""" - return docker_exec_python_json +def test_repo(weblate_ssh_pubkey: str) -> EphemeralGitHubRepo: + """Ephemeral GitHub repo with fixture docs and Weblate deploy key.""" + token = os.environ.get("GH_TEST_REPO_TOKEN", "").strip() + if not token: + pytest.skip( + "GH_TEST_REPO_TOKEN is not set in the job environment " + "(repository Actions secret with classic PAT 'repo' scope; " + "not available on pull_request workflows from forks)" + ) + + repo_name = default_repo_name() + manager = EphemeralGitHubRepo(token, repo_name) + try: + manager.create_repo() + manager.push_fixtures(FIXTURES_DIR, branch=TEST_BRANCH) + manager.add_deploy_key(weblate_ssh_pubkey) + yield manager + finally: + manager.delete_repo() diff --git a/tests/integration/lib/__init__.py b/tests/integration/lib/__init__.py index e69de29..6ac87f3 100644 --- a/tests/integration/lib/__init__.py +++ b/tests/integration/lib/__init__.py @@ -0,0 +1,5 @@ +# SPDX-FileCopyrightText: 2026 Andrew Zhang +# +# SPDX-License-Identifier: BSL-1.0 + +"""Shared helpers for integration tests.""" diff --git a/tests/integration/lib/docker_exec.py b/tests/integration/lib/docker_exec.py index 4978f81..71eff55 100644 --- a/tests/integration/lib/docker_exec.py +++ b/tests/integration/lib/docker_exec.py @@ -2,22 +2,18 @@ # # SPDX-License-Identifier: BSL-1.0 -"""Helper to execute Python snippets inside the running Weblate container.""" +"""Execute commands inside the running Weblate Docker container.""" from __future__ import annotations import json import os import subprocess +from typing import Any - -def _compose_cmd() -> list[str]: - compose_file = os.environ.get( - "WEBLATE_COMPOSE_FILE", - "docker/docker-compose.yml", - ) - project = os.environ.get("WEBLATE_COMPOSE_PROJECT", "cppa-weblate-plugin") - return ["docker", "compose", "-f", compose_file, "-p", project] +_COMPOSE_FILE = os.environ.get("WEBLATE_COMPOSE_FILE", "docker/docker-compose.yml") +_COMPOSE_PROJECT = os.environ.get("WEBLATE_COMPOSE_PROJECT", "cppa-weblate-plugin") +_PYTHON = "/app/venv/bin/python" def _weblate_django_preamble() -> str: @@ -30,35 +26,54 @@ def _weblate_django_preamble() -> str: ) -def docker_exec_python(snippet: str, *, timeout: float = 30.0) -> str: - """Run a Python snippet inside the weblate container and return stdout.""" - code = _weblate_django_preamble() + snippet - cmd = [ - *_compose_cmd(), - "exec", - "-T", - "weblate", - "/app/venv/bin/python", - "-c", - code, +def _compose_cmd(*args: str) -> list[str]: + return [ + "docker", + "compose", + "-f", + _COMPOSE_FILE, + "-p", + _COMPOSE_PROJECT, + *args, ] + + +def docker_exec_python(snippet: str, *, timeout: float = 120.0) -> str: + """Run a Python snippet in the weblate container; return stdout (stripped).""" + code = _weblate_django_preamble() + snippet result = subprocess.run( - cmd, + _compose_cmd("exec", "-T", "weblate", _PYTHON, "-c", code), + timeout=timeout, capture_output=True, text=True, - timeout=timeout, check=False, ) if result.returncode != 0: - raise RuntimeError( - f"docker exec failed (rc={result.returncode}):\n" - f"stdout: {result.stdout}\n" - f"stderr: {result.stderr}" + msg = ( + f"docker exec failed (exit {result.returncode}):\n" + f"stdout: {result.stdout}\nstderr: {result.stderr}" ) + raise RuntimeError(msg) return result.stdout.strip() -def docker_exec_python_json(snippet: str, *, timeout: float = 30.0) -> object: - """Run a Python snippet that prints JSON and return the parsed result.""" - raw = docker_exec_python(snippet, timeout=timeout) - return json.loads(raw) +def docker_exec_python_json(snippet: str, *, timeout: float = 120.0) -> Any: + """Run a Python snippet and parse stdout as JSON.""" + return json.loads(docker_exec_python(snippet, timeout=timeout)) + + +def docker_exec_read_file(path: str) -> str: + """Read a file from the weblate container.""" + result = subprocess.run( + _compose_cmd("exec", "-T", "weblate", "cat", path), + capture_output=True, + text=True, + check=False, + ) + if result.returncode != 0: + msg = ( + f"docker exec cat failed (exit {result.returncode}):\n" + f"stderr: {result.stderr}" + ) + raise RuntimeError(msg) + return result.stdout.strip() diff --git a/tests/integration/lib/gh_repo.py b/tests/integration/lib/gh_repo.py new file mode 100644 index 0000000..6ef8093 --- /dev/null +++ b/tests/integration/lib/gh_repo.py @@ -0,0 +1,182 @@ +# SPDX-FileCopyrightText: 2026 Andrew Zhang +# +# SPDX-License-Identifier: BSL-1.0 + +"""Ephemeral GitHub repository lifecycle for integration tests (stdlib only).""" + +from __future__ import annotations + +import base64 +import json +import os +import urllib.error +import urllib.request +from pathlib import Path +from typing import Any + +_GITHUB_API = "https://api.github.com" + + +def _api_error_message(method: str, path: str, code: int, raw: bytes) -> str: + body = raw.decode(errors="replace") + return f"GitHub API {method} {path} failed: {code} {body}" + + +class EphemeralGitHubRepo: + """Create, populate, and destroy a temporary GitHub repo for integration tests.""" + + __test__ = False # not a pytest test class + + def __init__(self, token: str, repo_name: str) -> None: + self.token = token + self.repo_name = repo_name + self.owner: str | None = None + self.repo_full_name: str | None = None + self._created = False + + def _request( + self, + method: str, + path: str, + *, + body: dict[str, Any] | None = None, + expected: tuple[int, ...] = (200, 201), + ) -> Any: + url = f"{_GITHUB_API}{path}" + data: bytes | None = None + headers = { + "Authorization": f"Bearer {self.token}", + "Accept": "application/vnd.github+json", + "X-GitHub-Api-Version": "2022-11-28", + } + if body is not None: + data = json.dumps(body).encode() + headers["Content-Type"] = "application/json" + + req = urllib.request.Request(url, data=data, headers=headers, method=method) + try: + with urllib.request.urlopen(req, timeout=60.0) as resp: + raw = resp.read() + code = resp.getcode() + except urllib.error.HTTPError as e: + raw = e.read() + code = e.code + if code not in expected: + raise RuntimeError(_api_error_message(method, path, code, raw)) from e + if not raw: + return None + return json.loads(raw.decode()) + + if code not in expected: + raise RuntimeError(_api_error_message(method, path, code, raw)) + if not raw: + return None + return json.loads(raw.decode()) + + def resolve_owner(self) -> str: + if self.owner: + return self.owner + user = self._request("GET", "/user") + assert isinstance(user, dict) + login = user.get("login") + if not login: + raise RuntimeError("GitHub /user did not return login") + self.owner = str(login) + return self.owner + + def create_repo(self) -> str: + """Create a repo; return SSH clone URL.""" + owner = self.resolve_owner() + self._request( + "POST", + "/user/repos", + body={ + "name": self.repo_name, + "private": False, + "auto_init": True, + }, + expected=(201,), + ) + self._created = True + self.repo_full_name = f"{owner}/{self.repo_name}" + return f"git@github.com:{self.repo_full_name}.git" + + def _put_file( + self, + path: str, + content: bytes, + branch: str, + *, + message: str, + ) -> None: + owner = self.resolve_owner() + encoded = base64.b64encode(content).decode("ascii") + api_path = f"/repos/{owner}/{self.repo_name}/contents/{path}" + self._request( + "PUT", + api_path, + body={ + "message": message, + "content": encoded, + "branch": branch, + }, + expected=(201,), + ) + + def push_fixtures(self, fixture_dir: Path, branch: str) -> None: + """Upload fixture files under doc/ on the given branch.""" + owner = self.resolve_owner() + # Create branch from default if needed via initial commit on branch + files = [ + ("doc/quickbook_fixture.qbk", "quickbook_fixture.qbk"), + ("doc/asciidoc_fixture.adoc", "asciidoc_fixture.adoc"), + ] + for dest, src_name in files: + src = fixture_dir / src_name + if not src.is_file(): + raise FileNotFoundError(src) + self._put_file( + dest, + src.read_bytes(), + branch, + message=f"Add {dest} for integration tests", + ) + # Ensure branch exists as default for clones + _ = owner + + def add_deploy_key(self, public_key: str, title: str = "weblate-ci") -> None: + """Register read-only deploy key on the repo.""" + owner = self.resolve_owner() + self._request( + "POST", + f"/repos/{owner}/{self.repo_name}/keys", + body={ + "title": title, + "key": public_key.strip(), + "read_only": True, + }, + expected=(201,), + ) + + def delete_repo(self) -> None: + """Delete the repository (no-op if never created).""" + if not self._created: + return + owner = self.resolve_owner() + try: + self._request( + "DELETE", + f"/repos/{owner}/{self.repo_name}", + expected=(204,), + ) + except RuntimeError: + pass + finally: + self._created = False + + +def default_repo_name() -> str: + """Unique repo name for this CI run.""" + run_id = os.environ.get("GITHUB_RUN_ID", os.environ.get("PYTEST_XDIST_WORKER", "")) + suffix = run_id or os.getpid() + return f"cppa-weblate-func-test-{suffix}" diff --git a/tests/integration/lib/http.py b/tests/integration/lib/http.py index 93c3a3a..648d5c8 100644 --- a/tests/integration/lib/http.py +++ b/tests/integration/lib/http.py @@ -17,6 +17,11 @@ def base_url() -> str: return os.environ.get("WEBLATE_LIVE_BASE_URL", "http://localhost:8080").rstrip("/") +def auth_header(token: str) -> str: + """Weblate API token auth (see Weblate 5.16 REST API docs).""" + return f"Token {token}" + + def http_json( method: str, path: str, @@ -29,7 +34,7 @@ def http_json( url = f"{base_url()}{path}" headers: dict[str, str] = {"Accept": "application/json"} if token is not None: - headers["Authorization"] = f"Bearer {token}" + headers["Authorization"] = auth_header(token) data: bytes | None = None if body is not None: diff --git a/tests/integration/lib/weblate_api.py b/tests/integration/lib/weblate_api.py new file mode 100644 index 0000000..4ab49d2 --- /dev/null +++ b/tests/integration/lib/weblate_api.py @@ -0,0 +1,563 @@ +# SPDX-FileCopyrightText: 2026 Andrew Zhang +# +# SPDX-License-Identifier: BSL-1.0 + +"""Weblate REST API client for integration tests (stdlib only).""" + +from __future__ import annotations + +import json +import mimetypes +import os +import time +import uuid +from pathlib import Path +from typing import Any +from urllib.error import HTTPError +from urllib.parse import urlparse +from urllib.request import Request, urlopen + +from tests.integration.lib.http import auth_header, base_url, http_json + +# Weblate blocks localhost/loopback in project.web (SSRF protection). +_DEFAULT_PROJECT_WEB = "https://example.com/" + + +def project_web_url(live_base_url: str | None = None) -> str: + """Return a project ``web`` URL accepted by Weblate in CI/local stacks.""" + override = os.environ.get("WEBLATE_PROJECT_WEB", "").strip() + if override: + return override if override.endswith("/") else f"{override}/" + + base = (live_base_url or base_url()).rstrip("/") + host = (urlparse(base).hostname or "").lower() + if host in {"localhost", "127.0.0.1", "::1"}: + return _DEFAULT_PROJECT_WEB + return f"{base}/" + + +def _expect_status( + code: int, allowed: tuple[int, ...], label: str, detail: Any +) -> None: + if code not in allowed: + raise AssertionError(f"{label} failed: {code} {detail}") + + +def _unwrap_api_result(body: dict[str, Any]) -> dict[str, Any]: + """Unwrap ``{"result": ...}`` envelopes from Weblate write endpoints.""" + result = body.get("result") + if isinstance(result, dict): + return result + return body + + +def component_defaults_payload( + *, + name: str, + slug: str, + file_format: str, + filemask: str, + template: str = "", + new_base: str | None = None, + repo: str = "local:", + vcs: str = "local", + source_language_code: str = "en", + language_regex: str = "", +) -> dict[str, Any]: + """Component fields aligned with ``BoostComponentService`` (services.py).""" + return { + "name": name, + "slug": slug, + "repo": repo, + "vcs": vcs, + "file_format": file_format, + "filemask": filemask, + "template": template, + "new_base": new_base or template, + "source_language": {"code": source_language_code}, + "edit_template": False, + "manage_units": False, + "license": "", + "allow_translation_propagation": False, + "enable_suggestions": True, + "suggestion_voting": False, + "suggestion_autoaccept": 0, + "check_flags": "", + "language_regex": language_regex, + } + + +def _payload_to_multipart_fields(payload: dict[str, Any]) -> dict[str, str]: + """Convert JSON API payload values to multipart form strings.""" + fields: dict[str, str] = {} + for key, value in payload.items(): + if key == "source_language" and isinstance(value, dict): + fields["source_language"] = str(value.get("code", "en")) + elif isinstance(value, bool): + fields[key] = "true" if value else "false" + else: + fields[key] = str(value) + return fields + + +def _multipart_encode( + fields: dict[str, str], files: dict[str, tuple[str, bytes, str]] +) -> tuple[bytes, str]: + """Build multipart/form-data body for file upload.""" + boundary = f"----WebKitFormBoundary{uuid.uuid4().hex}" + lines: list[bytes] = [] + + for name, value in fields.items(): + lines.append(f"--{boundary}\r\n".encode()) + lines.append(f'Content-Disposition: form-data; name="{name}"\r\n\r\n'.encode()) + lines.append(value.encode()) + lines.append(b"\r\n") + + for name, (filename, content, content_type) in files.items(): + lines.append(f"--{boundary}\r\n".encode()) + lines.append( + ( + f'Content-Disposition: form-data; name="{name}"; ' + f'filename="{filename}"\r\n' + ).encode() + ) + lines.append(f"Content-Type: {content_type}\r\n\r\n".encode()) + lines.append(content) + lines.append(b"\r\n") + + lines.append(f"--{boundary}--\r\n".encode()) + body = b"".join(lines) + content_type = f"multipart/form-data; boundary={boundary}" + return body, content_type + + +class WeblateAPI: + """Thin wrapper around Weblate's REST API for functional tests.""" + + def __init__(self, token: str, *, live_base_url: str | None = None) -> None: + self.token = token + self._base = (live_base_url or base_url()).rstrip("/") + + def _url(self, path: str) -> str: + if not path.startswith("/"): + path = f"/{path}" + return f"{self._base}{path}" + + def _api_path(self, url_or_path: str) -> str: + if url_or_path.startswith("http"): + return url_or_path.replace(self._base, "", 1) + if not url_or_path.startswith("/"): + return f"/{url_or_path}" + return url_or_path + + def create_project(self, name: str, slug: str | None = None) -> dict[str, Any]: + slug = slug or name.lower().replace(" ", "-") + code, body = http_json( + "POST", + "/api/projects/", + token=self.token, + body={ + "name": name, + "slug": slug, + "web": project_web_url(self._base), + }, + ) + _expect_status(code, (200, 201), "create_project", body) + assert isinstance(body, dict) + return body + + def create_component( + self, + project_slug: str, + *, + name: str, + slug: str, + file_format: str, + filemask: str, + template: str, + new_base: str | None = None, + repo: str = "local:", + vcs: str = "local", + language_regex: str = "", + ) -> dict[str, Any]: + payload = component_defaults_payload( + name=name, + slug=slug, + file_format=file_format, + filemask=filemask, + template=template, + new_base=new_base, + repo=repo, + vcs=vcs, + language_regex=language_regex, + ) + code, body = http_json( + "POST", + f"/api/projects/{project_slug}/components/", + token=self.token, + body=payload, + ) + _expect_status(code, (200, 201), "create_component", body) + assert isinstance(body, dict) + component = _unwrap_api_result(body) + task_url = component.get("task_url") + if isinstance(task_url, str) and task_url: + self.wait_for_background_task(task_url) + return component + + def _post_multipart( + self, + path: str, + fields: dict[str, str], + files: dict[str, tuple[str, bytes, str]], + *, + label: str, + timeout: float = 120.0, + ) -> tuple[int, Any]: + body_bytes, content_type = _multipart_encode(fields, files) + req = Request( + self._url(path), + data=body_bytes, + headers={ + "Authorization": auth_header(self.token), + "Content-Type": content_type, + }, + method="POST", + ) + try: + with urlopen(req, timeout=timeout) as resp: + raw = resp.read() + code = resp.getcode() + except HTTPError as e: + raw = e.read() + code = e.code + + if not raw: + return code, {"status_code": code} + try: + parsed: Any = json.loads(raw.decode()) + except json.JSONDecodeError: + parsed = raw.decode(errors="replace") + _expect_status(code, (200, 201), label, parsed) + return code, parsed + + def create_component_from_docfile( + self, + project_slug: str, + *, + name: str, + slug: str, + file_format: str, + docfile_path: Path, + filemask: str = "*.qbk", + language_regex: str = "", + ) -> dict[str, Any]: + """Create a component by uploading a document (Weblate multipart API).""" + content = docfile_path.read_bytes() + mime, _ = mimetypes.guess_type(str(docfile_path)) + mime = mime or "application/octet-stream" + payload = component_defaults_payload( + name=name, + slug=slug, + file_format=file_format, + filemask=filemask, + template="", + new_base="", + language_regex=language_regex, + ) + fields = _payload_to_multipart_fields(payload) + fields["new_lang"] = "add" + _code, body = self._post_multipart( + f"/api/projects/{project_slug}/components/", + fields, + {"docfile": (docfile_path.name, content, mime)}, + label="create_component_from_docfile", + ) + assert isinstance(body, dict) + component = _unwrap_api_result(body) + resolved_slug = str(component.get("slug", slug)) + task_url = component.get("task_url") + if isinstance(task_url, str) and task_url: + self.wait_for_background_task(task_url) + self.wait_for_component(project_slug, resolved_slug) + return component + + def wait_for_background_task( + self, + task_url: str, + *, + timeout: float = 120.0, + interval: float = 2.0, + ) -> dict[str, Any]: + """Poll ``GET /api/tasks/(uuid)/`` until component creation finishes.""" + path = self._api_path(task_url) + deadline = time.monotonic() + timeout + last: tuple[int, Any] = (0, None) + while time.monotonic() < deadline: + code, body = http_json("GET", path, token=self.token) + last = (code, body) + if code == 200 and isinstance(body, dict) and body.get("completed"): + return body + time.sleep(interval) + raise TimeoutError( + f"Background task {task_url} not completed after {timeout}s: " + f"{last[0]} {last[1]}" + ) + + def wait_for_component( + self, + project_slug: str, + component_slug: str, + *, + timeout: float = 120.0, + interval: float = 2.0, + ) -> dict[str, Any]: + """Poll until a component is visible (docfile create can be asynchronous).""" + deadline = time.monotonic() + timeout + last: tuple[int, Any] = (0, None) + while time.monotonic() < deadline: + code, body = http_json( + "GET", + f"/api/components/{project_slug}/{component_slug}/", + token=self.token, + ) + last = (code, body) + if code == 200 and isinstance(body, dict): + return body + time.sleep(interval) + raise TimeoutError( + f"Component {project_slug}/{component_slug} not ready after {timeout}s: " + f"{last[0]} {last[1]}" + ) + + def ensure_translation( + self, + project_slug: str, + component_slug: str, + language_code: str, + ) -> dict[str, Any]: + """Ensure a translation exists (idempotent; uses component-scoped API paths).""" + self.wait_for_component(project_slug, component_slug) + code, body = http_json( + "GET", + f"/api/components/{project_slug}/{component_slug}/translations/", + token=self.token, + ) + assert code == 200, f"list translations failed: {code} {body}" + assert isinstance(body, dict) + for item in body.get("results", []): + if not isinstance(item, dict): + continue + lang = item.get("language_code") + if lang is None: + lang_obj = item.get("language") + if isinstance(lang_obj, dict): + lang = lang_obj.get("code") + if lang == language_code: + return item + + code, body = http_json( + "POST", + f"/api/components/{project_slug}/{component_slug}/translations/", + token=self.token, + body={"language_code": language_code}, + ) + if code in (200, 201) and isinstance(body, dict): + if body.get("language_code"): + return body + result = body.get("result") + if isinstance(result, dict): + return result + return body + if code == 400 and isinstance(body, (dict, list)): + detail = json.dumps(body).lower() + if "already exists" in detail: + return {"language_code": language_code} + raise AssertionError(f"add translation failed: {code} {body}") + + def upload_file( + self, + project_slug: str, + component_slug: str, + language_code: str, + file_path: Path, + *, + method: str = "translate", + ) -> dict[str, Any]: + """Upload a translation file (multipart POST).""" + content = file_path.read_bytes() + mime, _ = mimetypes.guess_type(str(file_path)) + mime = mime or "application/octet-stream" + body_bytes, content_type = _multipart_encode( + {"method": method}, + {"file": (file_path.name, content, mime)}, + ) + url = self._url( + f"/api/translations/{project_slug}/{component_slug}/{language_code}/file/" + ) + req = Request( + url, + data=body_bytes, + headers={ + "Authorization": auth_header(self.token), + "Content-Type": content_type, + }, + method="POST", + ) + try: + with urlopen(req, timeout=120.0) as resp: + raw = resp.read() + code = resp.getcode() + except HTTPError as e: + raw = e.read() + code = e.code + + if not raw: + return {"status_code": code} + try: + parsed: Any = json.loads(raw.decode()) + except json.JSONDecodeError: + parsed = raw.decode(errors="replace") + _expect_status(code, (200, 201), "upload_file", parsed) + if isinstance(parsed, dict): + return parsed + return {"status_code": code, "body": parsed} + + def list_units( + self, + project_slug: str, + component_slug: str, + language_code: str, + *, + min_count: int = 1, + timeout: float = 120.0, + interval: float = 2.0, + ) -> list[dict[str, Any]]: + """List translation units, polling until strings are extracted.""" + path = ( + f"/api/translations/{project_slug}/{component_slug}/" + f"{language_code}/units/?page_size=100" + ) + deadline = time.monotonic() + timeout + last: tuple[int, Any] = (0, None) + while time.monotonic() < deadline: + code, body = http_json("GET", path, token=self.token) + last = (code, body) + if code != 200: + break + assert isinstance(body, dict) + results = body.get("results") + if isinstance(results, list) and len(results) >= min_count: + return results + time.sleep(interval) + raise AssertionError(f"list_units failed: {last[0]} {last[1]}") + + @staticmethod + def unit_api_url(unit: dict[str, Any]) -> str: + """Return the REST path for PATCHing a unit from a list-units result item.""" + url = unit.get("url") + if isinstance(url, str) and url: + return url + nested = unit.get("unit") + if isinstance(nested, str) and nested: + return nested + if isinstance(nested, dict): + nested_url = nested.get("url") + if isinstance(nested_url, str) and nested_url: + return nested_url + unit_id = unit.get("id") + if unit_id is not None: + return f"/api/units/{unit_id}/" + raise KeyError(f"unit has no API URL: {unit!r}") + + def submit_translation(self, unit_url: str, target: str) -> dict[str, Any]: + """PATCH a unit with translated target text (state 20 = translated).""" + path = self._api_path(unit_url) + code, body = http_json( + "PATCH", + path, + token=self.token, + body={"target": [target], "state": 20}, + ) + assert code == 200, f"submit_translation failed: {code} {body}" + assert isinstance(body, dict) + return body + + def download_file( + self, + project_slug: str, + component_slug: str, + language_code: str, + ) -> bytes: + url = self._url( + f"/api/translations/{project_slug}/{component_slug}/{language_code}/file/" + ) + req = Request( + url, + headers={"Authorization": auth_header(self.token)}, + method="GET", + ) + try: + with urlopen(req, timeout=120.0) as resp: + return resp.read() + except HTTPError as e: + raise AssertionError( + f"download_file failed: {e.code} {e.read().decode(errors='replace')}" + ) from e + + def poll_celery_task( + self, + task_id: str, + *, + timeout: float = 240.0, + interval: float = 3.0, + ) -> Any: + """Poll Celery task result inside the Weblate container.""" + from tests.integration.lib.docker_exec import docker_exec_python + + deadline = time.monotonic() + timeout + snippet_template = """ +import json +import time + +from celery.result import AsyncResult +from weblate.utils.celery import app + +task_id = {task_id!r} +deadline = time.monotonic() + {remaining:.1f} +result = None +while time.monotonic() < deadline: + ar = AsyncResult(task_id, app=app) + if ar.ready(): + if ar.failed(): + raise RuntimeError(str(ar.result)) + result = ar.result + break + time.sleep({interval}) +else: + raise TimeoutError(f"Task {{task_id}} not ready") + +print(json.dumps({{"ok": True, "result": result}})) +""" + last_exc: BaseException | None = None + while time.monotonic() < deadline: + remaining = max(deadline - time.monotonic(), interval) + snippet = snippet_template.format( + task_id=task_id, + remaining=remaining, + interval=interval, + ) + try: + out = docker_exec_python(snippet, timeout=timeout) + data = json.loads(out) + return data.get("result") + except (RuntimeError, json.JSONDecodeError, TimeoutError) as exc: + last_exc = exc + time.sleep(interval) + raise TimeoutError( + f"Celery task {task_id} did not complete within {timeout}s: {last_exc}" + ) + + @staticmethod + def unique_slug(prefix: str) -> str: + return f"{prefix}-{uuid.uuid4().hex[:8]}" diff --git a/tests/integration/test_functional.py b/tests/integration/test_functional.py new file mode 100644 index 0000000..9851cf1 --- /dev/null +++ b/tests/integration/test_functional.py @@ -0,0 +1,310 @@ +# SPDX-FileCopyrightText: 2026 Andrew Zhang +# +# SPDX-License-Identifier: BSL-1.0 + +"""P1 integration functional tests. + +Requires a live Weblate stack (Docker Compose) and optional GH_TEST_REPO_TOKEN +for add-or-update / BoostComponentService tests. +""" + +from __future__ import annotations + +import json +from dataclasses import dataclass + +import pytest + +from tests.integration.conftest import ( + FIXTURES_DIR, + TEST_BRANCH, + TEST_LANG_CODE, + TEST_VERSION, +) +from tests.integration.lib.gh_repo import EphemeralGitHubRepo +from tests.integration.lib.http import http_json +from tests.integration.lib.weblate_api import WeblateAPI + +pytestmark = pytest.mark.integration + +QBK_FIXTURE = FIXTURES_DIR / "quickbook_fixture.qbk" +KNOWN_SOURCE_STRING = "Complex QuickBook test fixture" +ZH_HANS_TRANSLATION = "复杂 QuickBook 测试夹具" + + +@dataclass(frozen=True) +class CreatedProjectComponent: + project_slug: str + component_slug: str + + +@pytest.fixture(scope="class") +def created_project_component(weblate_api: WeblateAPI) -> CreatedProjectComponent: + """Project + QuickBook component for round-trip tests.""" + project_slug = WeblateAPI.unique_slug("func-qbk") + component_slug = "qbk-fixture" + weblate_api.create_project("Functional QBK", project_slug) + # Docfile multipart upload (no empty local VCS template paths). + created = weblate_api.create_component_from_docfile( + project_slug, + name="QBK Fixture", + slug=component_slug, + file_format="quickbook", + docfile_path=QBK_FIXTURE, + filemask="doc/*.qbk", + language_regex=f"^{TEST_LANG_CODE}$", + ) + component_slug = str(created.get("slug", component_slug)) + weblate_api.ensure_translation(project_slug, component_slug, TEST_LANG_CODE) + return CreatedProjectComponent( + project_slug=project_slug, + component_slug=component_slug, + ) + + +# --------------------------------------------------------------------------- +# P1: QuickBook round-trip via Weblate REST API +# --------------------------------------------------------------------------- + + +class TestQuickBookRoundTrip: + """Upload QBK, translate a unit, download translated file.""" + + def test_units_extracted( + self, + weblate_api: WeblateAPI, + created_project_component: CreatedProjectComponent, + ) -> None: + units = weblate_api.list_units( + created_project_component.project_slug, + created_project_component.component_slug, + TEST_LANG_CODE, + ) + assert len(units) > 0 + sources = [ + u.get("source", [""])[0] if isinstance(u.get("source"), list) else "" + for u in units + ] + assert any(KNOWN_SOURCE_STRING in s for s in sources), sources[:5] + + def test_submit_translation( + self, + weblate_api: WeblateAPI, + created_project_component: CreatedProjectComponent, + ) -> None: + units = weblate_api.list_units( + created_project_component.project_slug, + created_project_component.component_slug, + TEST_LANG_CODE, + ) + match = next( + (u for u in units if KNOWN_SOURCE_STRING in str(u.get("source", ""))), + None, + ) + assert match is not None + unit_url = WeblateAPI.unit_api_url(match) + weblate_api.submit_translation(unit_url, ZH_HANS_TRANSLATION) + + def test_download_translated_qbk( + self, + weblate_api: WeblateAPI, + created_project_component: CreatedProjectComponent, + ) -> None: + raw = weblate_api.download_file( + created_project_component.project_slug, + created_project_component.component_slug, + TEST_LANG_CODE, + ) + text = raw.decode("utf-8", errors="replace") + assert ZH_HANS_TRANSLATION in text, "translated QBK content was not found" + + +# --------------------------------------------------------------------------- +# P1: BoostComponentService E2E (in-container, uses test repo) +# Run before TestAddOrUpdateCeleryFlow so process_all sees an empty DB. +# --------------------------------------------------------------------------- + + +class TestBoostComponentServiceE2E: + """Exercise BoostComponentService inside the Weblate container.""" + + @staticmethod + def _service_snippet( + test_repo: EphemeralGitHubRepo, + *, + run_process_all: bool = False, + run_twice: bool = False, + ) -> str: + owner = test_repo.resolve_owner() + repo_name = test_repo.repo_name + return f""" +import json +import os +import tempfile + +os.environ.setdefault("DJANGO_SETTINGS_MODULE", "weblate.settings_docker") +import django +django.setup() + +from weblate.auth.models import User +from weblate.trans.models import Component, Project +from boost_weblate.endpoint.services import BoostComponentService + +organization = {owner!r} +submodule = {repo_name!r} +lang_code = {TEST_LANG_CODE!r} +version = {TEST_VERSION!r} +branch = {TEST_BRANCH!r} + +user = User.objects.get(username="admin") +service = BoostComponentService( + organization=organization, + lang_code=lang_code, + version=version, + extensions=None, +) + +out = {{"clone_ok": False, "configs": [], "project_slug": None, "component_count": 0}} + +tmpdir = tempfile.mkdtemp() +try: + ok = service.clone_repository(submodule, tmpdir, branch) + out["clone_ok"] = bool(ok) + if ok: + configs = service.scan_documentation_files(tmpdir) + out["configs"] = [ + {{"name": c.get("component_name"), "format": c.get("file_format")}} + for c in configs + ] +finally: + import shutil + shutil.rmtree(tmpdir, ignore_errors=True) + +if {run_process_all!r}: + from boost_weblate.endpoint.services import _submodule_slug + slug = f"boost-{{_submodule_slug(submodule)}}-documentation-{{lang_code}}" + out["project_slug"] = slug + out["component_count_before"] = Component.objects.filter(project__slug=slug).count() + request = type("R", (), {{"user": user}})() + results = service.process_all([submodule], user=user, request=request) + out["process_all"] = results + out["component_count"] = Component.objects.filter(project__slug=slug).count() + + if {run_twice!r}: + count1 = out["component_count"] + service.process_all([submodule], user=user, request=request) + count2 = Component.objects.filter(project__slug=slug).count() + out["component_count_after_second"] = count2 + out["idempotent"] = count1 == count2 + +print(json.dumps(out)) +""" + + def test_clone_and_scan(self, exec_python, test_repo: EphemeralGitHubRepo) -> None: + out = json.loads( + exec_python(self._service_snippet(test_repo, run_process_all=False)) + ) + assert out["clone_ok"] is True + formats = {c["format"] for c in out["configs"]} + assert "quickbook" in formats + assert any(c["format"] == "asciidoc" for c in out["configs"]) + + def test_project_component_creation( + self, exec_python, test_repo: EphemeralGitHubRepo + ) -> None: + """Direct process_all on a DB with no prior components for this repo.""" + out = json.loads( + exec_python(self._service_snippet(test_repo, run_process_all=True)) + ) + assert out.get("project_slug") + assert out.get("component_count_before") == 0 + assert out["component_count"] > 0 + slug = out["project_slug"] + check = exec_python( + f""" +import os +os.environ.setdefault("DJANGO_SETTINGS_MODULE", "weblate.settings_docker") +import django +django.setup() +from weblate.trans.models import Project, Component +slug = {slug!r} +assert Project.objects.filter(slug=slug).exists() +assert Component.objects.filter(project__slug=slug).exists() +print("ok") +""" + ) + assert check == "ok" + + def test_idempotency(self, exec_python, test_repo: EphemeralGitHubRepo) -> None: + out = json.loads( + exec_python( + self._service_snippet(test_repo, run_process_all=True, run_twice=True) + ) + ) + assert out.get("idempotent") is True + + +# --------------------------------------------------------------------------- +# P1: add-or-update Celery flow (uses ephemeral GitHub test repo) +# Runs after E2E so HTTP/Celery path is tested against existing components too. +# --------------------------------------------------------------------------- + + +@pytest.fixture(scope="class") +def add_or_update_task(api_token: str, test_repo: EphemeralGitHubRepo) -> str: + """Accepted add-or-update request; returns Celery task id.""" + owner = test_repo.resolve_owner() + body = { + "organization": owner, + "version": TEST_VERSION, + "add_or_update": {TEST_LANG_CODE: [test_repo.repo_name]}, + } + code, data = http_json( + "POST", + "/boost-endpoint/add-or-update/", + token=api_token, + body=body, + ) + assert code == 202, f"expected 202: {code} {data}" + assert isinstance(data, dict) + assert data.get("status") == "accepted" + assert data.get("task_id") + return str(data["task_id"]) + + +class TestAddOrUpdateCeleryFlow: + """POST /boost-endpoint/add-or-update/ and poll Celery completion.""" + + def test_add_or_update_task_completes( + self, weblate_api: WeblateAPI, add_or_update_task: str + ) -> None: + result = weblate_api.poll_celery_task(add_or_update_task, timeout=300.0) + assert isinstance(result, dict) + lang_result = result.get(TEST_LANG_CODE) + assert lang_result is not None + # At least one submodule processed without fatal errors + assert isinstance(lang_result, list) + assert len(lang_result) > 0 + + def test_add_or_update_invalid_returns_400(self, api_token: str) -> None: + code, data = http_json( + "POST", + "/boost-endpoint/add-or-update/", + token=api_token, + body={"version": TEST_VERSION, "add_or_update": {TEST_LANG_CODE: ["x"]}}, + ) + assert code == 400, f"expected 400: {code} {data}" + assert isinstance(data, dict) + assert "errors" in data + + def test_add_or_update_unauthenticated_returns_401(self) -> None: + code, _ = http_json( + "POST", + "/boost-endpoint/add-or-update/", + body={ + "organization": "x", + "version": TEST_VERSION, + "add_or_update": {TEST_LANG_CODE: ["y"]}, + }, + ) + assert code in (401, 403) diff --git a/uv.lock b/uv.lock index c98bddd..92ec515 100644 --- a/uv.lock +++ b/uv.lock @@ -752,6 +752,10 @@ dev = [ {name = "coverage"}, {name = "pytest-cov"} ] +integration = [ + {name = "pytest"}, + {name = "pytest-timeout"} +] lint = [ {name = "prek"}, {name = "pytest"} @@ -780,6 +784,10 @@ dev = [ {name = "coverage", extras = ["toml"], specifier = ">=7.6.0"}, {name = "pytest-cov", specifier = ">=7.1.0"} ] +integration = [ + {name = "pytest", specifier = ">=8.3"}, + {name = "pytest-timeout", specifier = ">=2.3.1"} +] lint = [ {name = "prek", specifier = "==0.3.13"}, {name = "pytest", specifier = ">=8.3"} @@ -2751,6 +2759,18 @@ wheels = [ {url = "https://files.pythonhosted.org/packages/9d/7a/d968e294073affff457b041c2be9868a40c1c71f4a35fcc1e45e5493067b/pytest_cov-7.1.0-py3-none-any.whl", hash = "sha256:a0461110b7865f9a271aa1b51e516c9a95de9d696734a2f71e3e78f46e1d4678", size = 22876, upload-time = "2026-03-21T20:11:14.438Z"} ] +[[package]] +dependencies = [ + {name = "pytest"} +] +name = "pytest-timeout" +sdist = {url = "https://files.pythonhosted.org/packages/ac/82/4c9ecabab13363e72d880f2fb504c5f750433b2b6f16e99f4ec21ada284c/pytest_timeout-2.4.0.tar.gz", hash = "sha256:7e68e90b01f9eff71332b25001f85c75495fc4e3a836701876183c4bcfd0540a", size = 17973, upload-time = "2025-05-05T19:44:34.99Z"} +source = {registry = "https://pypi.org/simple"} +version = "2.4.0" +wheels = [ + {url = "https://files.pythonhosted.org/packages/fa/b6/3127540ecdf1464a00e5a01ee60a1b09175f6913f0644ac748494d9c4b21/pytest_timeout-2.4.0-py3-none-any.whl", hash = "sha256:c42667e5cdadb151aeb5b26d114aff6bdf5a907f176a007a30b940d3d865b5c2", size = 14382, upload-time = "2025-05-05T19:44:33.502Z"} +] + [[package]] name = "python-crontab" sdist = {url = "https://files.pythonhosted.org/packages/99/7f/c54fb7e70b59844526aa4ae321e927a167678660ab51dda979955eafb89a/python_crontab-3.3.0.tar.gz", hash = "sha256:007c8aee68dddf3e04ec4dce0fac124b93bd68be7470fc95d2a9617a15de291b", size = 57626, upload-time = "2025-07-13T20:05:35.535Z"}