From 55f00111cf86b65364ab36279bc104de466295ae Mon Sep 17 00:00:00 2001 From: Pablo <28235457+pablodz@users.noreply.github.com> Date: Wed, 4 Nov 2026 02:23:24 +0000 Subject: [PATCH 1/2] =?UTF-8?q?Implementa=20SDK=20Python=20independiente?= =?UTF-8?q?=20y=20publicaci=C3=B3n=20PyPI=20mediante=20OIDC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .ci/check_package.py | 96 +++++++++++++++ .ci/check_release.py | 18 +++ .ci/requirements.txt | 2 + .ci/sync_upstream.py | 85 +++++++++++++ .ci/test_release.py | 29 +++++ .github/workflows/ci.yml | 42 +++++++ .github/workflows/release.yml | 68 +++++++++++ .gitignore | 11 ++ MANIFEST.in | 2 + README.md | 47 ++++++- examples/offline.py | 10 ++ pyproject.toml | 19 +++ src/peru_sdk/__init__.py | 1 + src/peru_sdk/_http.py | 60 +++++++++ src/peru_sdk/errors.py | 9 ++ src/peru_sdk/reniec/__init__.py | 32 +++++ src/peru_sdk/reniec/models.py | 32 +++++ src/peru_sdk/sunat/__init__.py | 6 + src/peru_sdk/sunat/auth.py | 77 ++++++++++++ src/peru_sdk/sunat/models.py | 64 ++++++++++ src/peru_sdk/sunat/validity.py | 75 ++++++++++++ tests/data/conformance.json | 49 ++++++++ tests/test_sdk.py | 210 ++++++++++++++++++++++++++++++++ upstream.lock.json | 20 +++ 24 files changed, 1062 insertions(+), 2 deletions(-) create mode 100644 .ci/check_package.py create mode 100644 .ci/check_release.py create mode 100644 .ci/requirements.txt create mode 100644 .ci/sync_upstream.py create mode 100644 .ci/test_release.py create mode 100644 .github/workflows/ci.yml create mode 100644 .github/workflows/release.yml create mode 100644 .gitignore create mode 100644 MANIFEST.in create mode 100644 examples/offline.py create mode 100644 pyproject.toml create mode 100644 src/peru_sdk/__init__.py create mode 100644 src/peru_sdk/_http.py create mode 100644 src/peru_sdk/errors.py create mode 100644 src/peru_sdk/reniec/__init__.py create mode 100644 src/peru_sdk/reniec/models.py create mode 100644 src/peru_sdk/sunat/__init__.py create mode 100644 src/peru_sdk/sunat/auth.py create mode 100644 src/peru_sdk/sunat/models.py create mode 100644 src/peru_sdk/sunat/validity.py create mode 100644 tests/data/conformance.json create mode 100644 tests/test_sdk.py create mode 100644 upstream.lock.json diff --git a/.ci/check_package.py b/.ci/check_package.py new file mode 100644 index 0000000..c74cf42 --- /dev/null +++ b/.ci/check_package.py @@ -0,0 +1,96 @@ +"""Build, audit and test Python distributions in isolation from the source checkout.""" +import argparse +import email.parser +import hashlib +import json +import os +from pathlib import Path +import shutil +import subprocess +import sys +import tarfile +import tempfile +import tomllib +import zipfile + +ROOT = Path(__file__).resolve().parents[1] + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--output", type=Path, required=True) + args = parser.parse_args() + output = args.output.resolve() + dist, metadata = output / "dist", output / "manifest" + if output.exists() and any(output.iterdir()): + raise SystemExit("distribution destination must be empty") + dist.mkdir(parents=True) + metadata.mkdir() + Path("/var/tmp").mkdir(parents=True, exist_ok=True) + with tempfile.TemporaryDirectory(prefix="peru-python-package-", dir="/var/tmp") as directory: + task = Path(directory) + env = {**os.environ, "TMPDIR": directory, "TEMP": directory, "TMP": directory, + "PYTHONDONTWRITEBYTECODE": "1", "PIP_DISABLE_PIP_VERSION_CHECK": "1"} + + def run(*command, cwd=task, extra=None): + subprocess.run(command, cwd=cwd, env={**env, **(extra or {})}, check=True) + + source = task / "source" + shutil.copytree(ROOT, source, ignore=shutil.ignore_patterns(".git", "__pycache__", "*.egg-info", "build", "dist")) + project = tomllib.loads((source / "pyproject.toml").read_text())["project"] + run(sys.executable, "-m", "build", str(source), "--outdir", str(dist)) + wheel, = dist.glob("*.whl") + sdist, = dist.glob("*.tar.gz") + with zipfile.ZipFile(wheel) as archive: + names = archive.namelist() + if len(names) != len(set(names)): + raise SystemExit("duplicate wheel entries") + metadata_name, = [name for name in names if name.endswith(".dist-info/METADATA")] + info_root = metadata_name.rsplit("/", 1)[0] + "/" + for name in names: + if ".." in Path(name).parts or not name.startswith(("peru_sdk/", info_root)): + raise SystemExit(f"unexpected wheel file: {name}") + if name.endswith((".go", ".java", ".so", ".pyd", ".dll", ".whl", ".jar")): + raise SystemExit(f"foreign implementation or binary in wheel: {name}") + fields = email.parser.BytesParser().parsebytes(archive.read(metadata_name)) + if fields["Name"] != project["name"] or fields["Version"] != project["version"] or fields.get_all("Requires-Dist"): + raise SystemExit("unexpected package identity, version or runtime dependencies") + extracted = task / "sdist" + with tarfile.open(sdist) as archive: + for member in archive.getmembers(): + path = Path(member.name) + if path.is_absolute() or ".." in path.parts or not (member.isfile() or member.isdir()): + raise SystemExit("unsafe source archive entry") + target = extracted / path + if member.isdir(): + target.mkdir(parents=True, exist_ok=True) + else: + target.parent.mkdir(parents=True, exist_ok=True) + target.write_bytes(archive.extractfile(member).read()) + package_root, = extracted.iterdir() + if not (package_root / "tests/data/conformance.json").is_file(): + raise SystemExit("source distribution is missing its independent test corpus") + for path in package_root.rglob("*"): + if path.suffix in (".go", ".java", ".jar", ".whl") or path.name in ("go.mod", "sdk-go"): + raise SystemExit("foreign SDK in source distribution") + site = task / "installed" + run(sys.executable, "-m", "pip", "install", "--no-deps", "--no-cache-dir", "--only-binary=:all:", + "--target", str(site), str(wheel)) + extra = {"PYTHONPATH": str(site)} + run(sys.executable, "-c", "import peru_sdk; from pathlib import Path; assert Path(peru_sdk.__file__).is_relative_to(Path('installed').resolve())", extra=extra) + run(sys.executable, "-m", "unittest", "discover", "-s", str(package_root / "tests"), "-v", extra=extra) + run(sys.executable, "-m", "twine", "check", "--strict", str(wheel), str(sdist)) + commit = subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=ROOT, text=True).strip() + release = {"project": project["name"], "version": project["version"], "commit": commit, + "upstreams": json.loads((ROOT / "upstream.lock.json").read_text())} + (metadata / "release.json").write_text(json.dumps(release, indent=2) + "\n") + sums = [] + for path in sorted(output.rglob("*")): + if path.is_file(): + sums.append(f"{hashlib.sha256(path.read_bytes()).hexdigest()} {path.relative_to(output).as_posix()}") + (metadata / "SHA256SUMS").write_text("\n".join(sums) + "\n") + print("Verified isolated Python distributions:", output) + + +if __name__ == "__main__": + main() diff --git a/.ci/check_release.py b/.ci/check_release.py new file mode 100644 index 0000000..13f8034 --- /dev/null +++ b/.ci/check_release.py @@ -0,0 +1,18 @@ +"""Require a release tag to match the Python project version.""" +import os +from pathlib import Path +import tomllib + +ROOT = Path(__file__).resolve().parents[1] + + +def main(): + version = tomllib.loads((ROOT / "pyproject.toml").read_text())["project"]["version"] + expected = "refs/tags/v" + version + if os.environ.get("GITHUB_REF") != expected: + raise SystemExit(f"Publishing requires tag v{version}; project/tag mismatch") + print("Release tag matches project version:", version) + + +if __name__ == "__main__": + main() diff --git a/.ci/requirements.txt b/.ci/requirements.txt new file mode 100644 index 0000000..e8f6590 --- /dev/null +++ b/.ci/requirements.txt @@ -0,0 +1,2 @@ +build>=1.2,<2 +twine>=6,<8 diff --git a/.ci/sync_upstream.py b/.ci/sync_upstream.py new file mode 100644 index 0000000..ef6ff67 --- /dev/null +++ b/.ci/sync_upstream.py @@ -0,0 +1,85 @@ +"""Explicit, integrity-checked development synchronization; never used at runtime.""" +import argparse +import hashlib +import json +from pathlib import Path +import re +import subprocess +import sys +import tempfile +from urllib.request import urlopen + +ROOT = Path(__file__).resolve().parents[1] + + +def fetch(source: dict, path: str, digest: str) -> bytes: + repository, revision = source["repository"], source["revision"] + if not re.fullmatch(r"DataScienceResearchPeru/peru-sdk-(models|conformance)", repository): + raise ValueError("unexpected upstream repository") + if not re.fullmatch(r"[0-9a-f]{40}", revision) or not re.fullmatch(r"[0-9a-f]{64}", digest): + raise ValueError("upstreams require full commits and SHA-256 hashes") + relative = Path(path) + if relative.is_absolute() or ".." in relative.parts: + raise ValueError("invalid upstream path") + with urlopen(f"https://raw.githubusercontent.com/{repository}/{revision}/{path}", timeout=30) as response: + data = response.read(2 * 1024 * 1024 + 1) + if len(data) > 2 * 1024 * 1024 or hashlib.sha256(data).hexdigest() != digest: + raise ValueError(f"upstream integrity mismatch: {path}") + return data + + +def synchronize(check: bool): + lock = json.loads((ROOT / "upstream.lock.json").read_text(encoding="utf-8")) + if lock["format_version"] != 1: + raise ValueError("unsupported lock format") + case_path = Path(lock["cases_file"]) + if case_path.is_absolute() or ".." in case_path.parts: + raise ValueError("invalid corpus destination") + source = lock["conformance"] + corpus = fetch(source, source["path"], source["sha256"]) + outputs = {ROOT / case_path: corpus} + Path("/var/tmp").mkdir(parents=True, exist_ok=True) + with tempfile.TemporaryDirectory(prefix="peru-sdk-upstream-", dir="/var/tmp") as directory: + task = Path(directory) + if "generator" in lock: + language = lock["language"] + if language not in ("go", "python"): + raise ValueError("unsupported language") + generator = lock["generator"] + for name, digest in generator["files"].items(): + data = fetch(generator, name, digest) + target = task / "models" / name + target.parent.mkdir(parents=True, exist_ok=True) + target.write_bytes(data) + destination = task / "output" + subprocess.run([sys.executable, str(task / "models/tools/generate.py"), + "--language", language, "--output", str(destination)], check=True) + root = ROOT if language == "go" else ROOT / "src/peru_sdk" + for path in destination.rglob("*"): + if path.is_file(): + outputs[root / path.relative_to(destination)] = path.read_bytes() + differences = [] + for path, content in outputs.items(): + if check: + if not path.exists() or path.read_bytes() != content: + differences.append(str(path.relative_to(ROOT))) + else: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(content) + if differences: + raise ValueError("regeneration required: " + ", ".join(differences)) + print("Pinned upstreams and local outputs match." if check else "Updated from pinned upstreams.") + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--check", action="store_true") + args = parser.parse_args() + try: + synchronize(args.check) + except (ValueError, KeyError, OSError) as error: + parser.exit(1, str(error) + "\n") + + +if __name__ == "__main__": + main() diff --git a/.ci/test_release.py b/.ci/test_release.py new file mode 100644 index 0000000..32a728c --- /dev/null +++ b/.ci/test_release.py @@ -0,0 +1,29 @@ +import importlib.util +import os +from pathlib import Path +import unittest +from unittest.mock import patch + +SPEC = importlib.util.spec_from_file_location("release_check", Path(__file__).with_name("check_release.py")) +release = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(release) + + +class ReleaseGuardTests(unittest.TestCase): + def test_matching_version_tag(self): + with patch.dict(os.environ, {"GITHUB_REF": "refs/tags/v0.1.0a1"}): + release.main() + + def test_main_is_not_a_release(self): + with patch.dict(os.environ, {"GITHUB_REF": "refs/heads/main"}): + with self.assertRaises(SystemExit): + release.main() + + def test_tag_for_another_version_is_rejected(self): + with patch.dict(os.environ, {"GITHUB_REF": "refs/tags/v0.2.0"}): + with self.assertRaises(SystemExit): + release.main() + + +if __name__ == "__main__": + unittest.main() diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..acf6d3f --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,42 @@ +name: CI +on: + push: + branches: [main] + pull_request: + workflow_dispatch: +permissions: + contents: read +concurrency: + group: ci-${{ github.ref }} + cancel-in-progress: true +env: + PYTHONDONTWRITEBYTECODE: "1" +jobs: + upstream: + runs-on: ubuntu-latest + timeout-minutes: 10 + env: + TMPDIR: /var/tmp + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 + - uses: actions/setup-python@6d4fd22b20465ccc0f7354476588e8df7c9ca0e0 + with: + python-version: "3.14" + - run: python .ci/sync_upstream.py --check + native: + name: Python ${{ matrix.python }} / ${{ matrix.os }} + runs-on: ${{ matrix.os }} + timeout-minutes: 10 + strategy: + fail-fast: false + max-parallel: 2 + matrix: + python: ["3.11", "3.14"] + os: [ubuntu-latest, windows-latest] + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 + - uses: actions/setup-python@6d4fd22b20465ccc0f7354476588e8df7c9ca0e0 + with: + python-version: ${{ matrix.python }} + - run: python -m pip install . + - run: python -m unittest discover -s tests -v diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml new file mode 100644 index 0000000..de83e20 --- /dev/null +++ b/.github/workflows/release.yml @@ -0,0 +1,68 @@ +name: Release +on: + workflow_dispatch: + inputs: + publish: + description: Publish the verified version + type: boolean + default: false + pull_request: + push: + tags: ["v*"] +permissions: + contents: read +concurrency: + group: release-${{ github.ref }} + cancel-in-progress: false +jobs: + build: + runs-on: ubuntu-latest + timeout-minutes: 15 + env: + TMPDIR: /var/tmp + PYTHONDONTWRITEBYTECODE: "1" + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 + with: + persist-credentials: false + - uses: actions/setup-python@6d4fd22b20465ccc0f7354476588e8df7c9ca0e0 + with: + python-version: "3.14" + - run: python -m pip install -r .ci/requirements.txt + - run: python .ci/check_release.py + if: github.event_name == 'push' || inputs.publish + - run: python .ci/test_release.py + - run: python .ci/sync_upstream.py --check + - name: Build and test isolated distributions + run: | + stage=$(mktemp -d /var/tmp/peru-python-release.XXXXXXXX) + echo "SDK_DIST=$stage" >> "$GITHUB_ENV" + python .ci/check_package.py --output "$stage" + - uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 + with: + name: python-release-${{ github.sha }} + path: ${{ env.SDK_DIST }} + retention-days: 30 + publish: + if: github.event_name == 'push' || (github.event_name == 'workflow_dispatch' && inputs.publish) + needs: build + runs-on: ubuntu-latest + timeout-minutes: 10 + environment: pypi + permissions: + id-token: write + contents: read + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 + with: + persist-credentials: false + - uses: actions/download-artifact@9000827ccba6bdab643e8b6fd33ac0654aef8333 + with: + name: python-release-${{ github.sha }} + path: release + - name: Verify artifacts + run: sha256sum -c manifest/SHA256SUMS + working-directory: release + - uses: pypa/gh-action-pypi-publish@7245906fb61fa33a13554f9d5fb856fe61758086 + with: + packages-dir: release/dist/ diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..7d037a8 --- /dev/null +++ b/.gitignore @@ -0,0 +1,11 @@ +__pycache__/ +*.py[cod] +*.egg-info/ +build/ +dist/ +.venv/ +.env +*.pfx +*.p12 +*.pem +*.key diff --git a/MANIFEST.in b/MANIFEST.in new file mode 100644 index 0000000..317bb41 --- /dev/null +++ b/MANIFEST.in @@ -0,0 +1,2 @@ +include LICENSE README.md upstream.lock.json +recursive-include tests *.py *.json diff --git a/README.md b/README.md index 134603d..881554b 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,46 @@ -# Peru SDK python +# Peru SDK para Python -Componente del ecosistema [Peru SDK](https://github.com/DataScienceResearchPeru/peru-sdk). Licencia MIT. +SDK nativo Python 3.11+, licencia MIT y cero dependencias de runtime de terceros. Cliente HTTP de Consulta Integrada SUNAT y proveedor de identidad sintético. No implementa emisión ni un cliente oficial RENIEC; las consultas institucionales reales todavía no están verificadas. + +## Instalar y probar + +```bash +python3 -m pip install . +python3 examples/offline.py +python3 -m unittest discover -s tests -v +``` + +El ejemplo y los tests usan datos sintéticos, sin secretos ni consultas externas. El corpus está incluido en `tests/data/conformance.json`, incluso en la distribución fuente. La suite funciona sin otros repositorios del ecosistema. + +```python +from peru_sdk.sunat import StaticToken, ValidityClient, ValidityQuery + +# La aplicación proporciona el RUC consultante y un token autorizado. +client = ValidityClient(requester_ruc, StaticToken(access_token)) +query = ValidityQuery(issuer_ruc, "01", "F001", 123, "2026-11-02", "118.00") +result = client.check(query) # Esta llamada sí realiza una consulta HTTP. +``` + +`OAuth2TokenProvider(client_id, client_secret)` implementa la obtención y caché del token mediante client_credentials. No es el flujo de emisión con Clave SOL. Entrada inválida produce `ValueError`; errores institucionales y de protocolo producen `SDKError` con categoría y operación. Un estado de comprobante negativo no equivale a un error de transporte. + +## Modelos y conformidad + +`upstream.lock.json` fija por commit y SHA-256 el generador, los contratos y el corpus. Regeneración explícita: `python3 .ci/sync_upstream.py`; comprobación sin modificar archivos: `python3 .ci/sync_upstream.py --check`. El backend Python no requiere Go. El SDK instalado no descarga ni ejecuta el generador. + +## Empaquetado independiente + +```bash +python3 -m pip install -r .ci/requirements.txt +stage=$(mktemp -d /var/tmp/peru-python-dist.XXXXXXXX) +python3 .ci/check_package.py --output "$stage" +``` + +Se construyen wheel y sdist desde una copia aislada, se comprueban sus archivos y dependencias, se instala la wheel en un destino nuevo y se ejecuta la suite incluida en el sdist. `dist/` contiene exclusivamente los dos archivos publicables a PyPI; hashes y procedencia van en `manifest/`. + +## Publicar en PyPI + +El workflow Release produce artefactos en PR y ejecución manual. Publicar requiere una etiqueta que coincida exactamente con la versión de pyproject.toml: para 0.1.0a1, `v0.1.0a1`. Una ejecución manual tiene `publish=false` por defecto; para publicar manualmente, seleccionar la etiqueta y `publish=true`. + +Configurar una sola vez un Trusted Publisher en PyPI para el proyecto `peru-sdk` (sujeto a disponibilidad del nombre): owner `DataScienceResearchPeru`, repository `peru-sdk-python`, workflow `release.yml`, environment `pypi`. El job de publicación usa OIDC y solo recibe los dos archivos de `dist/`, después de verificar sus hashes. No requiere un token de PyPI permanente. + +No hay una publicación en PyPI completada por esta implementación. La configuración de cuenta y la publicación externa requieren ese alta. Las credenciales de otros SDKs no se usan en este repositorio. diff --git a/examples/offline.py b/examples/offline.py new file mode 100644 index 0000000..b86deed --- /dev/null +++ b/examples/offline.py @@ -0,0 +1,10 @@ +"""Run after installing sdk-python; uses synthetic data and makes no network calls.""" +from peru_sdk.reniec import Identity, IdentityQuery, MockIdentityService +from peru_sdk.sunat import parse_validity_response + +service = MockIdentityService([Identity("00000001", "PERSONA", "PRUEBA", "SINTETICA")]) +print("Identidad sintética:", service.lookup(IdentityQuery("00000001"))) +result = parse_validity_response( + b'{"success":true,"data":{"estadoCp":1,"estadoRuc":"00","condDomiRuc":"00"}}' +) +print("Estado de comprobante simulado:", result.document_status) diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..59508ad --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,19 @@ +[build-system] +requires = ["setuptools>=77"] +build-backend = "setuptools.build_meta" + +[project] +name = "peru-sdk" +version = "0.1.0a1" +description = "Clientes nativos para servicios del Estado peruano" +readme = "README.md" +requires-python = ">=3.11" +license = "MIT" +license-files = ["LICENSE"] +dependencies = [] + +[project.urls] +Repository = "https://github.com/DataScienceResearchPeru/peru-sdk-python" + +[tool.setuptools.packages.find] +where = ["src"] diff --git a/src/peru_sdk/__init__.py b/src/peru_sdk/__init__.py new file mode 100644 index 0000000..456f065 --- /dev/null +++ b/src/peru_sdk/__init__.py @@ -0,0 +1 @@ +"""Native clients for Peruvian government services.""" diff --git a/src/peru_sdk/_http.py b/src/peru_sdk/_http.py new file mode 100644 index 0000000..cafbfb2 --- /dev/null +++ b/src/peru_sdk/_http.py @@ -0,0 +1,60 @@ +import ipaddress +from urllib.error import HTTPError, URLError +from urllib.parse import urlsplit +from urllib.request import HTTPRedirectHandler, Request, build_opener + +from .errors import SDKError + +MAX_RESPONSE_BYTES = 1 << 20 + + +def checked_endpoint(value: str) -> str: + try: + parsed = urlsplit(value) + loopback = False + try: + loopback = ipaddress.ip_address(parsed.hostname or "").is_loopback + except ValueError: + pass + valid = ( + parsed.hostname and parsed.username is None and parsed.password is None + and not parsed.query and not parsed.fragment + and (parsed.scheme == "https" or (parsed.scheme == "http" and loopback)) + ) + parsed.port # Reject malformed ports before a request is made. + except ValueError: + valid = False + if not valid: + raise SDKError("configuration", "endpoint") + return value.rstrip("/") + + +class _NoRedirect(HTTPRedirectHandler): + def redirect_request(self, req, fp, code, msg, headers, newurl): + return None + + +class HTTPTransport: + """Bounded stdlib HTTP transport; redirects and automatic retries are disabled.""" + + def __init__(self, timeout: float = 10): + if timeout <= 0: + raise SDKError("configuration", "timeout") + self._timeout = timeout + self._opener = build_opener(_NoRedirect()) + + def post(self, url: str, headers: dict[str, str], body: bytes, operation: str) -> bytes: + request = Request(url, data=body, headers={"Accept": "application/json", **headers}, method="POST") + try: + with self._opener.open(request, timeout=self._timeout) as response: + result = response.read(MAX_RESPONSE_BYTES + 1) + if len(result) > MAX_RESPONSE_BYTES: + raise SDKError("protocol", operation) + return result + except HTTPError as exc: + status = exc.code + exc.close() + kind = {401: "authentication", 403: "authorization"}.get(status, "http") + raise SDKError(kind, operation, status=status) from None + except (URLError, OSError, ValueError): + raise SDKError("transport", operation) from None diff --git a/src/peru_sdk/errors.py b/src/peru_sdk/errors.py new file mode 100644 index 0000000..f29ff2c --- /dev/null +++ b/src/peru_sdk/errors.py @@ -0,0 +1,9 @@ +class SDKError(Exception): + """Stable category and operation; remote bodies and secrets are never formatted.""" + + def __init__(self, kind: str, operation: str, *, status: int = 0, code: str = ""): + self.kind = kind + self.operation = operation + self.status = status + self.code = code + super().__init__(f"{operation}: {kind} (HTTP {status})") diff --git a/src/peru_sdk/reniec/__init__.py b/src/peru_sdk/reniec/__init__.py new file mode 100644 index 0000000..3a13949 --- /dev/null +++ b/src/peru_sdk/reniec/__init__.py @@ -0,0 +1,32 @@ +"""Provisional identity interface. Only synthetic lookups are implemented.""" + +from typing import Protocol + +from ..errors import SDKError +from .models import Identity, IdentityQuery + + +class IdentityService(Protocol): + def lookup(self, query: IdentityQuery) -> Identity: ... + + +class MockIdentityService: + """An in-memory test provider, with no network access or official RENIEC adapter.""" + + def __init__(self, records: list[Identity]): + self._records: dict[str, Identity] = {} + for record in records: + record.__post_init__() + if record.dni in self._records: + raise ValueError("duplicate synthetic identity") + self._records[record.dni] = record + + def lookup(self, query: IdentityQuery) -> Identity: + query.__post_init__() + try: + return self._records[query.dni] + except KeyError: + raise SDKError("not_found", "synthetic_identity") from None + + +__all__ = ["Identity", "IdentityQuery", "IdentityService", "MockIdentityService"] diff --git a/src/peru_sdk/reniec/models.py b/src/peru_sdk/reniec/models.py new file mode 100644 index 0000000..3890cb6 --- /dev/null +++ b/src/peru_sdk/reniec/models.py @@ -0,0 +1,32 @@ +"""Contract models. Regenerate through the pinned models generator.""" +from dataclasses import dataclass +import re + +@dataclass(frozen=True) +class IdentityQuery: + dni: str + + def __post_init__(self) -> None: + if type(self.dni) is not str: + raise ValueError("dni: invalid type") + if re.fullmatch('^[0-9]{8}$', self.dni, flags=re.ASCII) is None: + raise ValueError("dni: invalid format") + +@dataclass(frozen=True) +class Identity: + dni: str + given_names: str + first_surname: str + second_surname: str + + def __post_init__(self) -> None: + if type(self.dni) is not str: + raise ValueError("dni: invalid type") + if re.fullmatch('^[0-9]{8}$', self.dni, flags=re.ASCII) is None: + raise ValueError("dni: invalid format") + if type(self.given_names) is not str: + raise ValueError("given_names: invalid type") + if type(self.first_surname) is not str: + raise ValueError("first_surname: invalid type") + if type(self.second_surname) is not str: + raise ValueError("second_surname: invalid type") diff --git a/src/peru_sdk/sunat/__init__.py b/src/peru_sdk/sunat/__init__.py new file mode 100644 index 0000000..eaeb99a --- /dev/null +++ b/src/peru_sdk/sunat/__init__.py @@ -0,0 +1,6 @@ +from .auth import OAuth2TokenProvider, StaticToken, TokenProvider +from .models import ValidityQuery, ValidityResult +from .validity import ValidityClient, parse_validity_response + +__all__ = ["OAuth2TokenProvider", "StaticToken", "TokenProvider", "ValidityQuery", + "ValidityResult", "ValidityClient", "parse_validity_response"] diff --git a/src/peru_sdk/sunat/auth.py b/src/peru_sdk/sunat/auth.py new file mode 100644 index 0000000..fbb9419 --- /dev/null +++ b/src/peru_sdk/sunat/auth.py @@ -0,0 +1,77 @@ +import json +from threading import Lock +from time import monotonic +from typing import Protocol +from urllib.parse import quote, urlencode + +from .._http import HTTPTransport, checked_endpoint +from ..errors import SDKError + +SCOPE = "https://api.sunat.gob.pe/v1/contribuyente/contribuyentes" + + +def valid_token(value: str) -> bool: + return type(value) is str and bool(value) and all(32 < ord(c) < 127 for c in value) + + +class TokenProvider(Protocol): + def token(self) -> str: ... + + +class StaticToken: + def __init__(self, token: str): + if not valid_token(token): + raise SDKError("configuration", "token") + self._value = token + + def __repr__(self) -> str: + return "StaticToken([redacted])" + + def token(self) -> str: + return self._value + + +class OAuth2TokenProvider: + """Consulta Integrada client_credentials, with expiry-aware in-memory caching.""" + + def __init__(self, client_id: str, client_secret: str, *, token_url: str | None = None, + transport: HTTPTransport | None = None): + if not client_id.strip() or not client_secret.strip(): + raise SDKError("configuration", "token") + self._client_id = client_id + self._secret = client_secret + self._url = token_url or ( + "https://api-seguridad.sunat.gob.pe/v1/clientesextranet/" + + quote(client_id, safe="") + "/oauth2/token/" + ) + checked_endpoint(self._url) + self._transport = transport or HTTPTransport() + self._lock = Lock() + self._access_token = "" + self._expires = 0.0 + self._clock = monotonic + + def __repr__(self) -> str: + return "OAuth2TokenProvider([redacted])" + + def token(self) -> str: + with self._lock: + if self._clock() < self._expires: + return self._access_token + started = self._clock() + form = urlencode({"grant_type": "client_credentials", "scope": SCOPE, + "client_id": self._client_id, "client_secret": self._secret}) + body = self._transport.post(self._url, {"Content-Type": "application/x-www-form-urlencoded"}, + form.encode("ascii"), "token") + try: + result = json.loads(body) + token, ttl, kind = result["access_token"], result["expires_in"], result["token_type"] + if not valid_token(token) or type(ttl) is not int or not 0 < ttl <= 86400 * 365: + raise ValueError + if type(kind) is not str or kind.lower() != "bearer": + raise ValueError + except (ValueError, KeyError, TypeError, UnicodeError): + raise SDKError("protocol", "token") from None + self._access_token = token + self._expires = started + ttl - min(30, ttl / 10) + return token diff --git a/src/peru_sdk/sunat/models.py b/src/peru_sdk/sunat/models.py new file mode 100644 index 0000000..2cea511 --- /dev/null +++ b/src/peru_sdk/sunat/models.py @@ -0,0 +1,64 @@ +"""Contract models. Regenerate through the pinned models generator.""" +from dataclasses import dataclass +from datetime import date +import re + +@dataclass(frozen=True) +class ValidityQuery: + issuer_ruc: str + document_type: str + series: str + number: int + issued_on: str + amount: str + + def __post_init__(self) -> None: + if type(self.issuer_ruc) is not str: + raise ValueError("issuer_ruc: invalid type") + if re.fullmatch('^[0-9]{11}$', self.issuer_ruc, flags=re.ASCII) is None: + raise ValueError("issuer_ruc: invalid format") + if type(self.document_type) is not str: + raise ValueError("document_type: invalid type") + if re.fullmatch('^[A-Z0-9]{2}$', self.document_type, flags=re.ASCII) is None: + raise ValueError("document_type: invalid format") + if type(self.series) is not str: + raise ValueError("series: invalid type") + if re.fullmatch('^[A-Z0-9]{4}$', self.series, flags=re.ASCII) is None: + raise ValueError("series: invalid format") + if type(self.number) is not int: + raise ValueError("number: invalid type") + if self.number < 1: + raise ValueError("number: out of range") + if self.number > 99999999: + raise ValueError("number: out of range") + if type(self.issued_on) is not str: + raise ValueError("issued_on: invalid type") + try: + parsed = date.fromisoformat(self.issued_on) + except ValueError: + raise ValueError("issued_on: invalid date") from None + if parsed.isoformat() != self.issued_on: + raise ValueError("issued_on: expected YYYY-MM-DD") + if type(self.amount) is not str: + raise ValueError("amount: invalid type") + if re.fullmatch('^(0|[1-9][0-9]{0,14})([.][0-9]{1,2})?$', self.amount, flags=re.ASCII) is None: + raise ValueError("amount: invalid format") + +@dataclass(frozen=True) +class ValidityResult: + document_status: str + issuer_status: str + issuer_address_status: str + observations: tuple[str, ...] + + def __post_init__(self) -> None: + if type(self.document_status) is not str: + raise ValueError("document_status: invalid type") + if re.fullmatch('^[0-9]+$', self.document_status, flags=re.ASCII) is None: + raise ValueError("document_status: invalid format") + if type(self.issuer_status) is not str: + raise ValueError("issuer_status: invalid type") + if type(self.issuer_address_status) is not str: + raise ValueError("issuer_address_status: invalid type") + if type(self.observations) is not tuple or any(type(v) is not str for v in self.observations): + raise ValueError("observations: invalid type") diff --git a/src/peru_sdk/sunat/validity.py b/src/peru_sdk/sunat/validity.py new file mode 100644 index 0000000..e46b611 --- /dev/null +++ b/src/peru_sdk/sunat/validity.py @@ -0,0 +1,75 @@ +from datetime import date +import json +import re + +from .._http import HTTPTransport, MAX_RESPONSE_BYTES, checked_endpoint +from ..errors import SDKError +from .auth import TokenProvider, valid_token +from .models import ValidityQuery, ValidityResult + + +def parse_validity_response(body: bytes) -> ValidityResult: + """Normalize status codes without treating query success as document acceptance.""" + try: + if len(body) > MAX_RESPONSE_BYTES: + raise ValueError + def reject_constant(value): + raise ValueError("non-JSON constant") + response = json.loads(body.decode("utf-8"), parse_constant=reject_constant) + if type(response) is not dict or type(response.get("success")) is not bool: + raise ValueError + if not response["success"]: + code = response.get("errorCode", "") + if type(code) is not str or re.fullmatch(r"[A-Za-z0-9_.-]{0,64}", code) is None: + code = "" + raise SDKError("remote", "validity", code=code) + data = response["data"] + if type(data) is not dict: + raise ValueError + + def read_code(name: str) -> str: + value = data[name] + if type(value) is int and value >= 0: + return str(value) + if type(value) is str and re.fullmatch(r"[0-9]+", value) is not None: + return value + raise ValueError + + observations = data.get("observaciones", data.get("Observaciones")) + if observations is None: + observations = [] + if type(observations) is not list or any(type(v) is not str for v in observations): + raise ValueError + return ValidityResult(read_code("estadoCp"), read_code("estadoRuc"), + read_code("condDomiRuc"), tuple(observations)) + except (ValueError, KeyError, TypeError, UnicodeError, OverflowError): + raise SDKError("protocol", "validity") from None + + +class ValidityClient: + """Read-only Consulta Integrada client. No implicit retries or global credentials.""" + + def __init__(self, requester_ruc: str, token_provider: TokenProvider, *, + base_url: str = "https://api.sunat.gob.pe", transport: HTTPTransport | None = None): + if type(requester_ruc) is not str or re.fullmatch(r"[0-9]{11}", requester_ruc) is None or token_provider is None: + raise SDKError("configuration", "validity") + self._url = (checked_endpoint(base_url) + "/v1/contribuyente/contribuyentes/" + + requester_ruc + "/validarcomprobante") + self._tokens = token_provider + self._transport = transport or HTTPTransport() + + def check(self, query: ValidityQuery) -> ValidityResult: + # Revalidate even when passed an instance created outside the normal constructor. + query.__post_init__() + data = {"numRuc": query.issuer_ruc, "codComp": query.document_type, + "numeroSerie": query.series, "numero": query.number, + "fechaEmision": date.fromisoformat(query.issued_on).strftime("%d/%m/%Y")} + # The contract permits only JSON decimal literals; never round through binary float. + body = (json.dumps(data, separators=(",", ":")).encode("utf-8")[:-1] + + b',"monto":' + query.amount.encode("ascii") + b"}") + token = self._tokens.token() + if not valid_token(token): + raise SDKError("configuration", "token") + response = self._transport.post(self._url, {"Content-Type": "application/json", + "Authorization": "Bearer " + token}, body, "validity") + return parse_validity_response(response) diff --git a/tests/data/conformance.json b/tests/data/conformance.json new file mode 100644 index 0000000..e9b2c09 --- /dev/null +++ b/tests/data/conformance.json @@ -0,0 +1,49 @@ +{ + "version": 1, + "provenance": "Casos sintéticos escritos para este proyecto a partir del manual oficial; ninguna respuesta proviene de una consulta real.", + "query": {"issuer_ruc": "20000000001", "document_type": "01", "series": "F001", "number": 123, "issued_on": "2026-11-02", "amount": "123456789012345.67"}, + "query_cases": [ + {"id": "valid", "patch": {}, "valid": true}, + {"id": "zero-amount", "patch": {"amount": "0.00"}, "valid": true}, + {"id": "unknown-document-code", "patch": {"document_type": "ZZ"}, "valid": true}, + {"id": "short-ruc", "patch": {"issuer_ruc": "123"}, "valid": false}, + {"id": "unicode-digits", "patch": {"issuer_ruc": "２０００００００００１"}, "valid": false}, + {"id": "invalid-day", "patch": {"issued_on": "2026-02-30"}, "valid": false}, + {"id": "non-iso-date", "patch": {"issued_on": "02/11/2026"}, "valid": false}, + {"id": "compact-date", "patch": {"issued_on": "20261102"}, "valid": false}, + {"id": "negative-amount", "patch": {"amount": "-1.00"}, "valid": false}, + {"id": "too-many-decimals", "patch": {"amount": "1.001"}, "valid": false}, + {"id": "exponent", "patch": {"amount": "1e2"}, "valid": false}, + {"id": "float-input", "patch": {"amount": 0.1}, "valid": false}, + {"id": "zero-number", "patch": {"number": 0}, "valid": false}, + {"id": "boolean-number", "patch": {"number": true}, "valid": false}, + {"id": "overflow-number", "patch": {"number": 100000000}, "valid": false}, + {"id": "lowercase-series", "patch": {"series": "f001"}, "valid": false}, + {"id": "injection-amount", "patch": {"amount": "0,\"extra\":1"}, "valid": false} + ], + "response_cases": [ + {"id": "accepted-integer", "body": {"success": true, "data": {"estadoCp": 1, "estadoRuc": "00", "condDomiRuc": "00"}}, "expected": {"document_status": "1", "issuer_status": "00", "issuer_address_status": "00", "observations": []}}, + {"id": "not-found-is-successful-query", "body": {"success": true, "data": {"estadoCp": "0", "estadoRuc": "00", "condDomiRuc": "00", "observaciones": []}}, "expected": {"document_status": "0", "issuer_status": "00", "issuer_address_status": "00", "observations": []}}, + {"id": "unknown-codes-preserved", "body": {"success": true, "future": 42, "data": {"estadoCp": 99, "estadoRuc": "98", "condDomiRuc": "97", "observaciones": ["Observación sintética"]}}, "expected": {"document_status": "99", "issuer_status": "98", "issuer_address_status": "97", "observations": ["Observación sintética"]}}, + {"id": "manual-capitalization", "body": {"success": true, "data": {"estadoCp": 2, "estadoRuc": "00", "condDomiRuc": "12", "Observaciones": ["Prueba"]}}, "expected": {"document_status": "2", "issuer_status": "00", "issuer_address_status": "12", "observations": ["Prueba"]}}, + {"id": "null-observations", "body": {"success": true, "data": {"estadoCp": 1, "estadoRuc": "00", "condDomiRuc": "00", "observaciones": null}}, "expected": {"document_status": "1", "issuer_status": "00", "issuer_address_status": "00", "observations": []}}, + {"id": "remote-error", "body": {"success": false, "errorCode": "0100", "message": "No registrar este texto"}, "error": "remote"}, + {"id": "missing-success", "body": {"data": {}}, "error": "protocol"}, + {"id": "string-success", "body": {"success": "true", "data": {}}, "error": "protocol"}, + {"id": "missing-data", "body": {"success": true}, "error": "protocol"}, + {"id": "null-data", "body": {"success": true, "data": null}, "error": "protocol"}, + {"id": "missing-status", "body": {"success": true, "data": {"estadoRuc": "00", "condDomiRuc": "00"}}, "error": "protocol"}, + {"id": "boolean-status", "body": {"success": true, "data": {"estadoCp": true, "estadoRuc": "00", "condDomiRuc": "00"}}, "error": "protocol"}, + {"id": "fractional-status", "body": {"success": true, "data": {"estadoCp": 1.5, "estadoRuc": "00", "condDomiRuc": "00"}}, "error": "protocol"}, + {"id": "null-status", "body": {"success": true, "data": {"estadoCp": null, "estadoRuc": "00", "condDomiRuc": "00"}}, "error": "protocol"}, + {"id": "invalid-observation", "body": {"success": true, "data": {"estadoCp": 1, "estadoRuc": "00", "condDomiRuc": "00", "observaciones": [null]}}, "error": "protocol"}, + {"id": "html-instead-of-json", "raw_body": "Servicio no disponible", "error": "protocol"}, + {"id": "truncated-json", "raw_body": "{\"success\":true,", "error": "protocol"} + ], + "identity_records": [{"dni": "00000001", "given_names": "PERSONA", "first_surname": "PRUEBA", "second_surname": "SINTETICA"}], + "identity_cases": [ + {"id": "synthetic-match", "dni": "00000001", "found": true}, + {"id": "synthetic-missing", "dni": "00000002", "found": false}, + {"id": "invalid-dni", "dni": "123", "invalid": true} + ] +} diff --git a/tests/test_sdk.py b/tests/test_sdk.py new file mode 100644 index 0000000..a1b3d5e --- /dev/null +++ b/tests/test_sdk.py @@ -0,0 +1,210 @@ +from concurrent.futures import ThreadPoolExecutor +from dataclasses import asdict +from decimal import Decimal +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +import json +from pathlib import Path +import threading +import unittest +from urllib.parse import parse_qs + +from peru_sdk._http import HTTPTransport, MAX_RESPONSE_BYTES +from peru_sdk.errors import SDKError +from peru_sdk.reniec import Identity, IdentityQuery, MockIdentityService +from peru_sdk.sunat import OAuth2TokenProvider, StaticToken, ValidityClient, ValidityQuery, parse_validity_response + +CASES = json.loads((Path(__file__).resolve().parent / "data/conformance.json").read_text(encoding="utf-8")) +ACCEPTED = b'{"success":true,"data":{"estadoCp":1,"estadoRuc":"00","condDomiRuc":"00"}}' + + +class Server: + def __init__(self, callback): + class Handler(BaseHTTPRequestHandler): + def do_POST(self): + # Consume the request before replying so close cannot reset an unread body. + self.request_body = self.rfile.read(int(self.headers.get("Content-Length", "0"))) + callback(self) + + def log_message(self, *args): + pass + + self.server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + self.thread = threading.Thread(target=self.server.serve_forever, daemon=True) + + def __enter__(self): + self.thread.start() + self.url = f"http://127.0.0.1:{self.server.server_port}" + return self + + def __exit__(self, *args): + self.server.shutdown() + self.server.server_close() + self.thread.join() + + +def respond(handler, body=ACCEPTED, status=200): + handler.send_response(status) + handler.send_header("Content-Type", "application/json") + handler.send_header("Content-Length", str(len(body))) + handler.end_headers() + handler.wfile.write(body) + + +class SharedConformance(unittest.TestCase): + def test_queries(self): + for case in CASES["query_cases"]: + with self.subTest(case=case["id"]): + data = {**CASES["query"], **case["patch"]} + if case["valid"]: + ValidityQuery(**data) + else: + with self.assertRaises(ValueError): + ValidityQuery(**data) + + def test_responses(self): + for case in CASES["response_cases"]: + with self.subTest(case=case["id"]): + body = case["raw_body"].encode() if "raw_body" in case else json.dumps(case["body"]).encode() + if "error" in case: + with self.assertRaises(SDKError) as caught: + parse_validity_response(body) + self.assertEqual(caught.exception.kind, case["error"]) + else: + result = json.loads(json.dumps(asdict(parse_validity_response(body)))) + self.assertEqual(result, case["expected"]) + + def test_synthetic_identities(self): + records = [Identity(**r) for r in CASES["identity_records"]] + service = MockIdentityService(records) + for case in CASES["identity_cases"]: + with self.subTest(case=case["id"]): + if case.get("invalid"): + with self.assertRaises(ValueError): + IdentityQuery(case["dni"]) + elif case["found"]: + self.assertEqual(service.lookup(IdentityQuery(case["dni"])), records[0]) + else: + with self.assertRaises(SDKError) as caught: + service.lookup(IdentityQuery(case["dni"])) + self.assertEqual(caught.exception.kind, "not_found") + with self.assertRaises(ValueError): + MockIdentityService(records + records) + + +class HTTPTests(unittest.TestCase): + def test_exchange_precision_and_expiry(self): + requests = [] + + def handle(handler): + body = handler.request_body + requests.append((handler.path, dict(handler.headers), body)) + if handler.path == "/token": + respond(handler, b'{"access_token":"synthetic-token","token_type":"Bearer","expires_in":100}') + else: + respond(handler) + + with Server(handle) as server: + tokens = OAuth2TokenProvider("synthetic-client", "synthetic-secret", token_url=server.url + "/token") + clock = [1000.0] + tokens._clock = lambda: clock[0] + client = ValidityClient("20000000002", tokens, base_url=server.url) + for _ in range(2): + self.assertEqual(client.check(ValidityQuery(**CASES["query"])).document_status, "1") + self.assertEqual(sum(path == "/token" for path, _, _ in requests), 1) + clock[0] += 91 + client.check(ValidityQuery(**CASES["query"])) + self.assertEqual(sum(path == "/token" for path, _, _ in requests), 2) + + for path, headers, body in requests: + if path == "/token": + form = parse_qs(body.decode()) + self.assertEqual(form["grant_type"], ["client_credentials"]) + self.assertEqual(form["scope"], ["https://api.sunat.gob.pe/v1/contribuyente/contribuyentes"]) + self.assertEqual(form["client_id"], ["synthetic-client"]) + self.assertEqual(form["client_secret"], ["synthetic-secret"]) + else: + self.assertEqual(path, "/v1/contribuyente/contribuyentes/20000000002/validarcomprobante") + self.assertEqual(headers["Authorization"], "Bearer synthetic-token") + payload = json.loads(body, parse_float=Decimal) + self.assertEqual(payload["numRuc"], "20000000001") + self.assertEqual(payload["fechaEmision"], "02/11/2026") + self.assertEqual(payload["monto"], Decimal("123456789012345.67")) + self.assertEqual(payload["numero"], 123) + + def test_concurrent_token_refresh(self): + requests = [] + + def handle(handler): + requests.append(handler.path) + respond(handler, b'{"access_token":"synthetic-token","token_type":"Bearer","expires_in":100}') + + with Server(handle) as server: + provider = OAuth2TokenProvider("id", "secret", token_url=server.url) + with ThreadPoolExecutor(max_workers=10) as executor: + self.assertEqual(list(executor.map(lambda _: provider.token(), range(10))), ["synthetic-token"] * 10) + self.assertEqual(len(requests), 1) + + def test_statuses_no_redirect_or_retry(self): + for status in (302, 401, 403, 429, 500): + with self.subTest(status=status): + requests = [] + + def handle(handler): + requests.append(handler.path) + handler.send_response(status) + handler.send_header("Location", "http://127.0.0.1:1/leak") + handler.end_headers() + handler.wfile.write(b"synthetic-secret") + + with Server(handle) as server: + client = ValidityClient("20000000002", StaticToken("token"), base_url=server.url) + with self.assertRaises(SDKError) as caught: + client.check(ValidityQuery(**CASES["query"])) + self.assertEqual(caught.exception.status, status) + self.assertNotIn("synthetic-secret", str(caught.exception)) + self.assertEqual(len(requests), 1) + + def test_response_limit(self): + with Server(lambda h: respond(h, b"x" * (MAX_RESPONSE_BYTES + 1))) as server: + client = ValidityClient("20000000002", StaticToken("token"), base_url=server.url) + with self.assertRaises(SDKError) as caught: + client.check(ValidityQuery(**CASES["query"])) + self.assertEqual(caught.exception.kind, "protocol") + + def test_invalid_endpoints_and_safe_repr(self): + for endpoint in ("http://example.com", "https://user:password@example.com", "https://example.com?token=secret"): + with self.assertRaises(SDKError): + ValidityClient("20000000002", StaticToken("token"), base_url=endpoint) + self.assertNotIn("do-not-log", repr(StaticToken("do-not-log"))) + self.assertNotIn("do-not-log", repr(OAuth2TokenProvider("id", "do-not-log"))) + with self.assertRaises(SDKError): + StaticToken("header\ninjection") + + def test_malformed_token_responses(self): + for body in ({}, None, {"access_token": "x", "token_type": "Bearer", "expires_in": 0}, + {"access_token": "x", "token_type": "Bearer", "expires_in": True}, + {"access_token": "x", "token_type": "Other", "expires_in": 10}): + with self.subTest(body=body): + with Server(lambda h: respond(h, json.dumps(body).encode())) as server: + provider = OAuth2TokenProvider("id", "secret", token_url=server.url) + with self.assertRaises(SDKError) as caught: + provider.token() + self.assertEqual(caught.exception.kind, "protocol") + + def test_validation_precedes_token_or_network(self): + class UnexpectedToken: + def token(self): + raise AssertionError("token requested before validation") + + query = ValidityQuery(**CASES["query"]) + object.__setattr__(query, "amount", "1e2") + client = ValidityClient("20000000002", UnexpectedToken()) + with self.assertRaises(ValueError): + client.check(query) + + def test_invalid_timeout(self): + self.assertRaises(SDKError, HTTPTransport, 0) + + +if __name__ == "__main__": + unittest.main() diff --git a/upstream.lock.json b/upstream.lock.json new file mode 100644 index 0000000..c73181a --- /dev/null +++ b/upstream.lock.json @@ -0,0 +1,20 @@ +{ + "format_version": 1, + "language": "python", + "generator": { + "repository": "DataScienceResearchPeru/peru-sdk-models", + "revision": "fbd07a48d26a9094014c94f0dfe2de2cdd0718d7", + "files": { + "tools/generate.py": "4cac653a5c3383fe07ed1f87dc16d0152d094fb8fa28ff88540a98e617376f32", + "contracts/sunat/validity.schema.json": "da1cc4837ae075fa6e3d4f3f7449f30a5d124582cd12397c0475c63f19fda2d9", + "contracts/reniec/identity.prototype.schema.json": "55419fe3c3d5d8d7a880fa11405be8edf7dc4c3eb49848f60b07fad640e07661" + } + }, + "conformance": { + "repository": "DataScienceResearchPeru/peru-sdk-conformance", + "revision": "96f6b934f2269865d9a524154537f0a579617083", + "path": "cases.json", + "sha256": "77d947fcb161b763413c85469d05e0a33d4162c4aebd8636c4890d490200d751" + }, + "cases_file": "tests/data/conformance.json" +} From 9d93bd023bdc188e22a9616684dfc87cbc691607 Mon Sep 17 00:00:00 2001 From: Pablo <28235457+pablodz@users.noreply.github.com> Date: Wed, 4 Nov 2026 02:30:32 +0000 Subject: [PATCH 2/2] =?UTF-8?q?Comprueba=20releases=20con=20la=20versi?= =?UTF-8?q?=C3=B3n=20vigente=20del=20paquete?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .ci/test_release.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/.ci/test_release.py b/.ci/test_release.py index 32a728c..4761a32 100644 --- a/.ci/test_release.py +++ b/.ci/test_release.py @@ -1,17 +1,19 @@ import importlib.util import os from pathlib import Path +import tomllib import unittest from unittest.mock import patch SPEC = importlib.util.spec_from_file_location("release_check", Path(__file__).with_name("check_release.py")) release = importlib.util.module_from_spec(SPEC) SPEC.loader.exec_module(release) +VERSION = tomllib.loads((release.ROOT / "pyproject.toml").read_text())["project"]["version"] class ReleaseGuardTests(unittest.TestCase): def test_matching_version_tag(self): - with patch.dict(os.environ, {"GITHUB_REF": "refs/tags/v0.1.0a1"}): + with patch.dict(os.environ, {"GITHUB_REF": "refs/tags/v" + VERSION}): release.main() def test_main_is_not_a_release(self): @@ -20,7 +22,8 @@ def test_main_is_not_a_release(self): release.main() def test_tag_for_another_version_is_rejected(self): - with patch.dict(os.environ, {"GITHUB_REF": "refs/tags/v0.2.0"}): + other_version = "0.0.0" if VERSION != "0.0.0" else "0.0.1" + with patch.dict(os.environ, {"GITHUB_REF": "refs/tags/v" + other_version}): with self.assertRaises(SystemExit): release.main()