commit bdd7a791e4a63a520142f0a4daca2ff8a2e60dee Author: Walter Oggioni Date: Sun Aug 2 15:25:25 2026 +0000 Implement kaya-rbcs: simplified RBCS clone on the Kaya framework - KayaApp with GET/PUT recursive routes backed by memcached (aiomcache) - env-based configuration, metadata-aware value encoding, key prefix/digest - Granian RSGI launcher (rbcs-server / python -m kaya_rbcs) - pinned requirements via pip-compile, alpine:3.24 Dockerfile with healthcheck - offline tests using an in-process fake memcached - Gitea Actions workflow to build and push the multi-arch Docker image diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..aaaf12b --- /dev/null +++ b/.dockerignore @@ -0,0 +1,12 @@ +.git +.gitignore +.venv +**/__pycache__ +*.pyc +.mypy_cache +build +dist +*.egg-info +tests +Dockerfile +.dockerignore diff --git a/.gitea/workflows/build.yaml b/.gitea/workflows/build.yaml new file mode 100644 index 0000000..d01165f --- /dev/null +++ b/.gitea/workflows/build.yaml @@ -0,0 +1,38 @@ +name: CI +on: + push: + tags: + - 'release/*' + +jobs: + build_and_push_docker_image: + name: "Build and push docker image" + runs-on: hostinger + steps: + - name: Checkout sources + uses: actions/checkout@v6 + - name: Set up Docker Buildx + uses: docker/setup-buildx-action@v3 + - name: Login to Gitea container registry + uses: docker/login-action@v3 + with: + registry: gitea.woggioni.net + username: woggioni-opencode-agent + password: ${{ secrets.PUBLISHER_TOKEN }} + - name: Extract metadata for image tags + id: meta + uses: docker/metadata-action@v5 + with: + images: gitea.woggioni.net/woggioni-opencode-agent/kaya-rbcs + flavor: latest=false + tags: | + type=match,pattern=release/(.*),group=1 + type=raw,value=latest + - name: Build and push docker image + uses: docker/build-push-action@v6 + with: + context: . + platforms: linux/amd64,linux/arm64 + push: true + pull: true + tags: ${{ steps.meta.outputs.tags }} diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..f43f878 --- /dev/null +++ b/.gitignore @@ -0,0 +1,7 @@ +.venv/ +__pycache__/ +*.pyc +.mypy_cache/ +build/ +dist/ +*.egg-info/ diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..056b84e --- /dev/null +++ b/Dockerfile @@ -0,0 +1,37 @@ +FROM alpine:3.24 + +ENV PYTHONDONTWRITEBYTECODE=1 \ + PYTHONUNBUFFERED=1 \ + PIP_DISABLE_PIP_VERSION_CHECK=1 \ + PIP_NO_CACHE_DIR=1 \ + RBCS_HOST=0.0.0.0 \ + RBCS_PORT=8080 \ + RBCS_MEMCACHE_HOST=memcached \ + RBCS_MEMCACHE_PORT=11211 + +RUN apk add --no-cache python3 + +WORKDIR /opt/kaya-rbcs + +COPY requirements.txt ./ +RUN python3 -m venv /opt/venv \ + && /opt/venv/bin/pip install -r requirements.txt + +COPY pyproject.toml README.md ./ +COPY src ./src +RUN /opt/venv/bin/pip install --no-deps . + +RUN addgroup -S rbcs \ + && adduser -S -G rbcs rbcs \ + && chown -R rbcs:rbcs /opt/kaya-rbcs /opt/venv + +ENV PATH="/opt/venv/bin:${PATH}" + +USER rbcs + +EXPOSE 8080 + +HEALTHCHECK --interval=30s --timeout=5s --start-period=10s --retries=3 \ + CMD python3 -c "import socket, os; socket.create_connection(('127.0.0.1', int(os.environ['RBCS_PORT'])), timeout=3).close()" + +ENTRYPOINT ["rbcs-server"] diff --git a/README.md b/README.md new file mode 100644 index 0000000..b73bcc8 --- /dev/null +++ b/README.md @@ -0,0 +1,87 @@ +# kaya-rbcs + +A simplified Python clone of [RBCS](https://gitea.woggioni.net/woggioni/rbcs) +(Remote Build Cache Server) built on the [Kaya](https://gitea.woggioni.net/woggioni/kaya) +web framework. + +It implements only the two core cache endpoints — `GET` and `PUT` — persisting +values to memcached. Authentication, RBAC, rate limiting, TRACE health checks, +compression and TLS are intentionally left out. + +## Behavior + +The cache key is the request path relative to a configurable URL prefix +(default `/`), normalized (`..` segments are resolved). Values are stored in +memcached together with the `Content-Type` and `Content-Disposition` metadata +sent on upload, mirroring RBCS' `CacheValueMetadata`. + +| Request | Response | +|------------------------|-----------------------------------------------------------------| +| `PUT /` | `201 Created`, body = key, `Content-Type: text/plain` | +| `GET /` (hit) | `200 OK`, body = value, stored `Content-Type` / `Content-Disposition` | +| `GET /` (miss) | `404 Not Found`, empty body | +| `GET/PUT` outside prefix | `400 Bad Request` | + +## Installation + +```bash +pip install -e . # + kaya-core, aiomcache +pip install -e .[server] # additionally installs granian + kaya-rsgi +``` + +Requires Python 3.10+ and a running memcached. + +## Running + +Start the server with the `rbcs-server` console script or `python -m kaya_rbcs`: + +```bash +rbcs-server +``` + +Configuration is read from environment variables: + +| Variable | Default | Description | +|-------------------------|---------------|-------------------------------------| +| `RBCS_HOST` | `127.0.0.1` | Bind address | +| `RBCS_PORT` | `8080` | Bind port | +| `RBCS_PATH_PREFIX` | `/` | URL prefix that maps to the cache | +| `RBCS_MEMCACHE_HOST` | `127.0.0.1` | Memcached host | +| `RBCS_MEMCACHE_PORT` | `11211` | Memcached port | +| `RBCS_KEY_PREFIX` | *(unset)* | String appended to each cache key | +| `RBCS_DIGEST` | *(unset)* | Hash the key with this algorithm (e.g. `sha256`) | +| `RBCS_MAX_AGE` | `7d` | Value TTL (`s`/`m`/`h`/`d`) | + +The app is also a plain ASGI application, so it can be served by any ASGI +server, e.g. `daphne kaya_rbcs.app:app`. + +### Usage example + +```bash +curl -X PUT -H 'Content-Type: application/octet-stream' \ + --data-binary @build-output.tar http://localhost:8080/my-module/build +curl http://localhost:8080/my-module/build -o build-output.tar +``` + +## Tests + +```bash +python -m unittest discover -s tests +``` + +Tests use an in-process fake memcached (text protocol), so they run offline +with no external dependencies. + +## Layout + +``` +src/kaya_rbcs/ +├── __init__.py # package exports + `rbcs-server` entry point +├── __main__.py # Granian (RSGI) launcher +├── config.py # env-based configuration +├── store.py # MemcacheStore: value+metadata encoding, key processing +└── app.py # create_app(): KayaApp with GET/PUT recursive routes +tests/ +├── fake_memcached.py +└── test_app.py +``` diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..ef3680f --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,53 @@ +[build-system] +requires = ["setuptools>=61.0"] +build-backend = "setuptools.build_meta" + +[project] +name = "kaya-rbcs" +dynamic = ["version"] +description = "Simplified clone of RBCS (Remote Build Cache Server) built on the Kaya web framework" +readme = "README.md" +requires-python = ">=3.10" +license = { text = "MIT" } +classifiers = [ + 'Development Status :: 3 - Alpha', + 'Topic :: Utilities', + 'Intended Audience :: Developers', + 'Environment :: Console', + 'Programming Language :: Python :: 3', +] + +dependencies = [ + "kaya-core", + "aiomcache", +] + +[project.optional-dependencies] +server = [ + "kaya-rsgi", + "granian", +] +dev = [ + "httpx", + "mypy", +] + +[project.scripts] +rbcs-server = "kaya_rbcs:main" + +[tool.setuptools.packages.find] +where = ["src"] +namespaces = true + +[tool.mypy] +python_version = "3.12" +disallow_untyped_defs = true +show_error_codes = true +no_implicit_optional = true +warn_return_any = true +warn_unused_ignores = true +strict = true +exclude = ["tests"] + +[tool.setuptools.dynamic] +version = { attr = "kaya_rbcs.__version__" } diff --git a/requirements.in b/requirements.in new file mode 100644 index 0000000..44756fe --- /dev/null +++ b/requirements.in @@ -0,0 +1,6 @@ +# runtime dependencies of kaya-rbcs +# (see pyproject.toml: [project] dependencies + [project.optional-dependencies] server) +kaya-core +aiomcache +kaya-rsgi +granian diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..c8b0f17 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,31 @@ +# +# This file is autogenerated by pip-compile with Python 3.14 +# by the following command: +# +# pip-compile --extra-index-url=https://pypi.org/simple --index-url=https://gitea.woggioni.net/api/packages/woggioni/pypi/simple --output-file=requirements.txt requirements.in +# +--index-url https://gitea.woggioni.net/api/packages/woggioni/pypi/simple +--extra-index-url https://pypi.org/simple + +aiomcache==0.8.2 + # via -r requirements.in +click==8.4.2 + # via granian +granian==2.7.9 + # via + # -r requirements.in + # kaya-rsgi +kaya-core==0.0.1 + # via + # -r requirements.in + # kaya-rsgi +kaya-rsgi==0.0.1 + # via -r requirements.in +pwo==0.1.2 + # via + # kaya-core + # kaya-rsgi +typing-extensions==4.16.0 + # via + # kaya-core + # pwo diff --git a/src/kaya_rbcs/__init__.py b/src/kaya_rbcs/__init__.py new file mode 100644 index 0000000..d1a992d --- /dev/null +++ b/src/kaya_rbcs/__init__.py @@ -0,0 +1,20 @@ +from .app import app, create_app +from .config import Config +from .store import CacheValueMetadata, MemcacheStore + +__version__ = '0.1.0' + + +def main() -> None: + from kaya_rbcs.__main__ import main as _main + + _main() + + +__all__ = [ + 'app', + 'create_app', + 'Config', + 'CacheValueMetadata', + 'MemcacheStore', +] diff --git a/src/kaya_rbcs/__main__.py b/src/kaya_rbcs/__main__.py new file mode 100644 index 0000000..d347220 --- /dev/null +++ b/src/kaya_rbcs/__main__.py @@ -0,0 +1,32 @@ +import sys +from typing import List, NoReturn + +from .config import Config + + +def main() -> NoReturn: + try: + from granian.cli import entrypoint + except ImportError as exc: + raise SystemExit( + "Granian is required to run the server, " + "install it with 'pip install kaya-rbcs[server]'" + ) from exc + + config = Config.from_env() + args: List[str] = [ + '--interface', + 'rsgi', + '--host', + config.host, + '--port', + str(config.port), + 'kaya_rbcs.app:app', + ] + sys.argv = [sys.argv[0], *args] + entrypoint() # type: ignore[no-untyped-call] + raise SystemExit(0) + + +if __name__ == '__main__': + main() diff --git a/src/kaya_rbcs/app.py b/src/kaya_rbcs/app.py new file mode 100644 index 0000000..c8b2b27 --- /dev/null +++ b/src/kaya_rbcs/app.py @@ -0,0 +1,108 @@ +from logging import getLogger +from posixpath import normpath +from typing import List, Mapping, Optional, Sequence + +from kaya.core import HttpContext, KayaApp + +from .config import Config +from .store import CacheValueMetadata, MemcacheStore + +log = getLogger(__name__) + +_APPLICATION_OCTET_STREAM = 'application/octet-stream' +_TEXT_PLAIN = 'text/plain' + + +def _normalize_prefix(prefix: str) -> str: + stripped = prefix.strip('/') + return '/' + stripped if stripped else '/' + + +def _route_pattern(prefix: str) -> str: + stripped = prefix.strip('/') + return '/*' if not stripped else '/' + stripped + '/*' + + +def key_from_path(path: str, prefix: str) -> Optional[str]: + """Normalize ``path`` and compute the cache key relative to ``prefix``. + + Returns ``None`` when the path does not fall under the prefix (or resolves + to an empty key), mirroring the 400 responses of the original RBCS for + unhandled / escaping paths. + """ + normalized = normpath(path) + prefix = _normalize_prefix(prefix) + if prefix == '/': + key = normalized.lstrip('/') + else: + if normalized != prefix and not normalized.startswith(prefix + '/'): + return None + key = normalized[len(prefix):].lstrip('/') + return key or None + + +def _first_header(headers: Mapping[str, Sequence[str]], name: str) -> Optional[str]: + values = headers.get(name) + if values is None: + return None + if isinstance(values, str): + return values + return values[0] if len(values) > 0 else None + + +def create_app(config: Config) -> KayaApp: + store = MemcacheStore( + host=config.memcache_host, + port=config.memcache_port, + key_prefix=config.key_prefix, + digest=config.digest, + max_age=config.max_age, + ) + prefix = config.path_prefix + app = KayaApp() + app.store = store # type: ignore[attr-defined] + + @app.GET(_route_pattern(prefix), recursive=True) + async def get_handler(ctx: HttpContext, path: List[str]) -> None: + key = key_from_path(ctx.path, prefix) + if key is None: + log.warning('Got GET request for unhandled path %s', ctx.path) + await ctx.send_empty(400) + return + result = await store.get(key) + if result is None: + log.debug('Cache miss for key %s', key) + await ctx.send_empty(404) + return + metadata, value = result + log.debug('Cache hit for key %s', key) + headers: dict[str, str] = { + 'content-type': metadata.content_type or _APPLICATION_OCTET_STREAM, + } + if metadata.content_disposition is not None: + headers['content-disposition'] = metadata.content_disposition + await ctx.send_bytes(200, value, headers) + + @app.PUT(_route_pattern(prefix), recursive=True) + async def put_handler(ctx: HttpContext, path: List[str]) -> None: + key = key_from_path(ctx.path, prefix) + if key is None: + log.warning('Got PUT request for unhandled path %s', ctx.path) + await ctx.send_empty(400) + return + chunks: List[bytes] = [] + async for chunk in ctx.request_body: + chunks.append(chunk) + value = b''.join(chunks) + metadata = CacheValueMetadata( + content_type=_first_header(ctx.headers, 'content-type'), + content_disposition=_first_header(ctx.headers, 'content-disposition'), + ) + await store.put(key, value, metadata) + log.debug('Added value for key %s', key) + await ctx.send_str(201, key, {'content-type': _TEXT_PLAIN}) + + return app + + +app = create_app(Config.from_env()) diff --git a/src/kaya_rbcs/config.py b/src/kaya_rbcs/config.py new file mode 100644 index 0000000..1ec4b74 --- /dev/null +++ b/src/kaya_rbcs/config.py @@ -0,0 +1,44 @@ +from dataclasses import dataclass, field +from datetime import timedelta +from os import getenv + + +def _parse_max_age(raw: str) -> timedelta: + if raw.endswith('s'): + return timedelta(seconds=int(raw[:-1])) + if raw.endswith('m'): + return timedelta(minutes=int(raw[:-1])) + if raw.endswith('h'): + return timedelta(hours=int(raw[:-1])) + if raw.endswith('d'): + return timedelta(days=int(raw[:-1])) + return timedelta(seconds=int(raw)) + + +@dataclass(frozen=True) +class Config: + host: str = '127.0.0.1' + port: int = 8080 + path_prefix: str = '/' + memcache_host: str = '127.0.0.1' + memcache_port: int = 11211 + key_prefix: str | None = None + digest: str | None = None + max_age: timedelta = field(default_factory=lambda: timedelta(days=7)) + + @staticmethod + def from_env() -> 'Config': + def get_int(name: str, default: int) -> int: + raw = getenv(name) + return int(raw) if raw is not None else default + + return Config( + host=getenv('RBCS_HOST', '127.0.0.1'), + port=get_int('RBCS_PORT', 8080), + path_prefix=getenv('RBCS_PATH_PREFIX', '/'), + memcache_host=getenv('RBCS_MEMCACHE_HOST', '127.0.0.1'), + memcache_port=get_int('RBCS_MEMCACHE_PORT', 11211), + key_prefix=getenv('RBCS_KEY_PREFIX'), + digest=getenv('RBCS_DIGEST'), + max_age=_parse_max_age(getenv('RBCS_MAX_AGE', '7d')), + ) diff --git a/src/kaya_rbcs/store.py b/src/kaya_rbcs/store.py new file mode 100644 index 0000000..3557429 --- /dev/null +++ b/src/kaya_rbcs/store.py @@ -0,0 +1,117 @@ +import hashlib +import json +import struct +from dataclasses import asdict, dataclass +from datetime import timedelta +from time import time +from typing import Callable, Optional, Tuple + +import aiomcache + +_THIRTY_DAYS = 30 * 24 * 60 * 60 + + +def _exptime(max_age: timedelta, clock: Callable[[], float]) -> int: + """Convert a relative TTL to a memcached ``exptime`` value. + + Memcached interprets ``exptime`` values larger than 30 days as absolute + Unix timestamps, so large TTLs must be converted explicitly. + """ + seconds = int(max_age.total_seconds()) + if seconds > _THIRTY_DAYS: + return int(clock()) + seconds + return seconds + + +@dataclass(frozen=True) +class CacheValueMetadata: + """Metadata stored alongside a cached value, mirroring RBCS' metadata.""" + + content_type: Optional[str] = None + content_disposition: Optional[str] = None + + +def process_cache_key(key: str, key_prefix: Optional[str], digest: Optional[str]) -> bytes: + """Compute the memcached key, mirroring RBCS' ``RBCS.processCacheKey``. + + The key prefix is appended to the key (as in the original implementation) + and, if a digest algorithm is configured, the result is hashed and returned + as a hexadecimal string so that it stays a valid memcached key. + """ + prefixed_key = key + key_prefix if key_prefix is not None else key + data = prefixed_key.encode('utf-8') + if digest is not None: + return hashlib.new(digest, data).hexdigest().encode('utf-8') + return data + + +def encode_value(value: bytes, metadata: CacheValueMetadata) -> bytes: + """Encode a cached value as ``[4-byte metadata length][json metadata][value]``.""" + metadata_bytes = json.dumps(asdict(metadata)).encode('utf-8') + return struct.pack('>I', len(metadata_bytes)) + metadata_bytes + value + + +def decode_value(payload: bytes) -> Tuple[CacheValueMetadata, bytes]: + """Decode a value previously encoded with :func:`encode_value`.""" + (size,) = struct.unpack('>I', payload[:4]) + metadata = json.loads(payload[4:4 + size].decode('utf-8')) + return CacheValueMetadata(**metadata), payload[4 + size:] + + +class MemcacheStore: + """Memcached-backed cache store with RBCS-compatible value encoding. + + The underlying ``aiomcache.Client`` is created lazily on first use so that + it is always bound to the currently running event loop; call :meth:`start` + eagerly inside a running loop if you want to fail fast, and :meth:`close` + to release the pool. + """ + + def __init__( + self, + host: str = '127.0.0.1', + port: int = 11211, + key_prefix: Optional[str] = None, + digest: Optional[str] = None, + max_age: timedelta = timedelta(days=7), + clock: Callable[[], float] = time, + ) -> None: + self._host = host + self._port = port + self._key_prefix = key_prefix + self._digest = digest + self._max_age = max_age + self._clock = clock + self._client: Optional[aiomcache.Client] = None + + def _ensure_client(self) -> aiomcache.Client: + if self._client is None: + self._client = aiomcache.Client(self._host, self._port) + return self._client + + async def start(self) -> None: + self._ensure_client() + + def _memcache_key(self, key: str) -> bytes: + return process_cache_key(key, self._key_prefix, self._digest) + + async def get(self, key: str) -> Optional[Tuple[CacheValueMetadata, bytes]]: + client = self._ensure_client() + payload = await client.get(self._memcache_key(key)) + if payload is None: + return None + return decode_value(payload) + + async def put(self, key: str, value: bytes, metadata: CacheValueMetadata) -> None: + client = self._ensure_client() + exptime = _exptime(self._max_age, self._clock) + await client.set( + self._memcache_key(key), + encode_value(value, metadata), + exptime=exptime, + ) + + async def close(self) -> None: + if self._client is not None: + await self._client.close() + self._client = None diff --git a/tests/fake_memcached.py b/tests/fake_memcached.py new file mode 100644 index 0000000..3f7cbd0 --- /dev/null +++ b/tests/fake_memcached.py @@ -0,0 +1,144 @@ +"""A minimal in-process memcached server (text protocol) for offline tests. + +It implements the subset of the memcached text protocol used by +``aiomcache`` (and thus by this project): ``get``, ``set``/``add``/ +``replace``/``append``/``prepend``, ``touch``, ``delete``, ``version``, +``flush_all`` and ``quit``, including lazy expiration. +""" + +import asyncio +from time import time +from typing import Dict, List, Optional, Tuple + +_THIRTY_DAYS = 30 * 24 * 60 * 60 + + +class FakeMemcachedServer: + def __init__(self) -> None: + self._data: Dict[bytes, Tuple[int, bytes]] = {} + self._server: Optional[asyncio.AbstractServer] = None + self.port: int = 0 + + async def start(self) -> None: + self._server = await asyncio.start_server(self._handle, '127.0.0.1', 0) + assert self._server.sockets is not None + self.port = self._server.sockets[0].getsockname()[1] + + async def stop(self) -> None: + if self._server is not None: + self._server.close() + await self._server.wait_closed() + + @staticmethod + def _expiry_timestamp(exptime: int, now: float) -> int: + if exptime == 0: + return 0 + if exptime > _THIRTY_DAYS: + return exptime + return int(now) + exptime + + def _expired(self, key: bytes, now: float) -> bool: + entry = self._data.get(key) + if entry is None: + return False + expiry, _ = entry + return 0 < expiry <= now + + async def _handle(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None: + try: + while True: + line = await reader.readline() + if not line: + break + parts = line.rstrip(b'\r\n').split() + if not parts: + continue + command = parts[0] + if command == b'get': + self._handle_get(parts[1:], writer) + elif command in (b'set', b'add', b'replace', b'append', b'prepend'): + await self._handle_store(command, parts, reader, writer) + elif command == b'touch': + self._handle_touch(parts, writer) + elif command == b'delete': + self._handle_delete(parts, writer) + elif command == b'version': + writer.write(b'VERSION 1.6.23\r\n') + elif command == b'flush_all': + self._data.clear() + writer.write(b'OK\r\n') + elif command == b'quit': + break + else: + writer.write(b'ERROR\r\n') + await writer.drain() + finally: + writer.close() + await writer.wait_closed() + + def _handle_get(self, keys: List[bytes], writer: asyncio.StreamWriter) -> None: + now = time() + for key in keys: + if key in self._data and not self._expired(key, now): + _, value = self._data[key] + writer.write( + b'VALUE ' + key + b' 0 ' + str(len(value)).encode('utf-8') + b'\r\n' + ) + writer.write(value + b'\r\n') + writer.write(b'END\r\n') + + async def _handle_store( + self, + command: bytes, + parts: List[bytes], + reader: asyncio.StreamReader, + writer: asyncio.StreamWriter, + ) -> None: + key = parts[1] + exptime = int(parts[3]) + length = int(parts[4]) + value = (await reader.readexactly(length + 2))[:-2] + now = time() + stored = False + if command == b'set': + self._data[key] = (self._expiry_timestamp(exptime, now), value) + stored = True + elif command == b'add': + if not (key in self._data and not self._expired(key, now)): + self._data[key] = (self._expiry_timestamp(exptime, now), value) + stored = True + elif command == b'replace': + if key in self._data and not self._expired(key, now): + self._data[key] = (self._expiry_timestamp(exptime, now), value) + stored = True + elif command in (b'append', b'prepend'): + if key in self._data and not self._expired(key, now): + expiry, existing = self._data[key] + self._data[key] = ( + expiry, + existing + value if command == b'append' else value + existing, + ) + stored = True + writer.write(b'STORED\r\n' if stored else b'NOT_STORED\r\n') + + def _handle_touch(self, parts: List[bytes], writer: asyncio.StreamWriter) -> None: + key = parts[1] + exptime = int(parts[2]) + if key in self._data and not self._expired(key, time()): + _, value = self._data[key] + self._data[key] = (self._expiry_timestamp(exptime, time()), value) + writer.write(b'TOUCHED\r\n') + else: + writer.write(b'NOT_FOUND\r\n') + + def _handle_delete(self, parts: List[bytes], writer: asyncio.StreamWriter) -> None: + key = parts[1] + if key in self._data: + del self._data[key] + writer.write(b'DELETED\r\n') + else: + writer.write(b'NOT_FOUND\r\n') + + @property + def data(self) -> Dict[bytes, Tuple[int, bytes]]: + return self._data diff --git a/tests/test_app.py b/tests/test_app.py new file mode 100644 index 0000000..7276a81 --- /dev/null +++ b/tests/test_app.py @@ -0,0 +1,208 @@ +import asyncio +import hashlib +import unittest +from datetime import timedelta +from typing import Any + +import httpx + +from kaya_rbcs.app import create_app, key_from_path +from kaya_rbcs.config import Config +from kaya_rbcs.store import MemcacheStore + +from fake_memcached import FakeMemcachedServer + + +class RbcsTest(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self) -> None: + self.memcache = FakeMemcachedServer() + await self.memcache.start() + self.app = create_app(self._config()) + self.store: MemcacheStore = getattr(self.app, 'store') + self.client = httpx.AsyncClient( + transport=httpx.ASGITransport(app=self.app), + base_url='http://testserver', + ) + + async def asyncTearDown(self) -> None: + await self.client.aclose() + await self.store.close() + await self.memcache.stop() + + def _config(self, **overrides: Any) -> Config: + params: dict[str, Any] = { + 'host': '127.0.0.1', + 'port': 8080, + 'path_prefix': '/', + 'memcache_host': '127.0.0.1', + 'memcache_port': self.memcache.port, + } + params.update(overrides) + return Config(**params) + + async def test_put_get_roundtrip(self) -> None: + key = 'abc123' + value = b'hello world' + put = await self.client.put( + '/' + key, + content=value, + headers={'content-type': 'application/octet-stream'}, + ) + self.assertEqual(201, put.status_code) + self.assertEqual(key, put.text) + self.assertEqual('text/plain', put.headers['content-type']) + + get = await self.client.get('/' + key) + self.assertEqual(200, get.status_code) + self.assertEqual(value, get.content) + self.assertEqual('application/octet-stream', get.headers['content-type']) + + async def test_get_missing_key(self) -> None: + get = await self.client.get('/does/not/exist') + self.assertEqual(404, get.status_code) + self.assertEqual(b'', get.content) + + async def test_nested_key(self) -> None: + value = b'nested value' + put = await self.client.put('/a/b/c', content=value) + self.assertEqual(201, put.status_code) + get = await self.client.get('/a/b/c') + self.assertEqual(200, get.status_code) + self.assertEqual(value, get.content) + + async def test_default_content_type(self) -> None: + await self.client.put('/no-type', content=b'x') + get = await self.client.get('/no-type') + self.assertEqual('application/octet-stream', get.headers['content-type']) + + async def test_content_disposition_roundtrip(self) -> None: + disposition = 'inline; filename="page.html"' + await self.client.put( + '/page', + content=b'', + headers={'content-type': 'text/html', 'content-disposition': disposition}, + ) + get = await self.client.get('/page') + self.assertEqual(200, get.status_code) + self.assertEqual('text/html', get.headers['content-type']) + self.assertEqual(disposition, get.headers['content-disposition']) + + async def test_key_prefix_appended(self) -> None: + app = create_app(self._config(key_prefix='suffix')) + store: MemcacheStore = getattr(app, 'store') + client = httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), + base_url='http://testserver', + ) + try: + put = await client.put('/key1', content=b'x') + self.assertEqual(201, put.status_code) + self.assertIn(b'key1suffix', self.memcache.data) + + get = await client.get('/key1') + self.assertEqual(200, get.status_code) + self.assertEqual(b'x', get.content) + finally: + await client.aclose() + await store.close() + + async def test_digest_hashes_key(self) -> None: + app = create_app(self._config(digest='sha256')) + store: MemcacheStore = getattr(app, 'store') + client = httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), + base_url='http://testserver', + ) + try: + put = await client.put('/key1', content=b'x') + self.assertEqual(201, put.status_code) + expected = hashlib.sha256(b'key1').hexdigest().encode('utf-8') + self.assertIn(expected, self.memcache.data) + + get = await client.get('/key1') + self.assertEqual(200, get.status_code) + self.assertEqual(b'x', get.content) + finally: + await client.aclose() + await store.close() + + async def test_path_prefix(self) -> None: + app = create_app(self._config(path_prefix='/cache')) + store: MemcacheStore = getattr(app, 'store') + client = httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), + base_url='http://testserver', + ) + try: + await client.put('/cache/entry', content=b'value') + get = await client.get('/cache/entry') + self.assertEqual(200, get.status_code) + self.assertEqual(b'value', get.content) + self.assertIn(b'entry', self.memcache.data) + finally: + await client.aclose() + await store.close() + + def test_key_from_path_rejects_escaping(self) -> None: + self.assertIsNone(key_from_path('/cache/../../etc/passwd', '/cache')) + self.assertIsNone(key_from_path('/cache/../', '/cache')) + self.assertIsNone(key_from_path('/', '/')) + self.assertEqual('a/b', key_from_path('/cache/a/b', '/cache')) + self.assertEqual('foo', key_from_path('/foo', '/')) + self.assertEqual('foo', key_from_path('/cache/foo', '/cache')) + + async def test_escaped_path_is_rejected(self) -> None: + # ASGI servers (and httpx) normalize dot segments in the request path + # before it reaches the app, so the route is never matched. Either way + # the request is rejected; a raw, unnormalized escaping path would be + # handled by key_from_path and rejected with 400. + app = create_app(self._config(path_prefix='/cache')) + store: MemcacheStore = getattr(app, 'store') + client = httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), + base_url='http://testserver', + ) + try: + get = await client.get('/cache/../../etc/passwd') + self.assertIn(get.status_code, (400, 404)) + put = await client.put('/cache/../../etc/passwd', content=b'x') + self.assertIn(put.status_code, (400, 404)) + finally: + await client.aclose() + await store.close() + + async def test_max_age_expiry(self) -> None: + app = create_app(self._config(max_age=timedelta(seconds=1))) + store: MemcacheStore = getattr(app, 'store') + client = httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), + base_url='http://testserver', + ) + try: + put = await client.put('/ttl', content=b'x') + self.assertEqual(201, put.status_code) + + get = await client.get('/ttl') + self.assertEqual(200, get.status_code) + + await asyncio.sleep(1.2) + get = await client.get('/ttl') + self.assertEqual(404, get.status_code) + finally: + await client.aclose() + await store.close() + + async def test_overwrite_value(self) -> None: + await self.client.put('/same', content=b'first') + await self.client.put('/same', content=b'second') + get = await self.client.get('/same') + self.assertEqual(200, get.status_code) + self.assertEqual(b'second', get.content) + + async def test_invalid_method_returns_404(self) -> None: + post = await self.client.post('/anything', content=b'x') + self.assertEqual(404, post.status_code) + + +if __name__ == '__main__': + unittest.main()