Reinitialize repo without reference solution

Author: daciwolf

.github/pull_request_template.md

@@ -0,0 +1,10 @@
+## Submission Checklist
+
+- [ ] I changed `solution.py`.
+- [ ] I did not edit the official tests to inflate my score.
+- [ ] I ran `python score.py`.
+- [ ] I added or updated my own tests under `tests/student/` if needed.
+
+## Notes
+
+Describe any tradeoffs, assumptions, or known gaps here.

.github/workflows/reference-tests.yml

@@ -0,0 +1,29 @@
+name: Challenge Smoke Test
+
+on:
+  push:
+  pull_request:
+
+jobs:
+  smoke-test:
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: Check out repository
+        uses: actions/checkout@v4
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.11"
+
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          python -m pip install -r requirements.txt
+
+      - name: Smoke test starter import
+        run: python -c "import solution"
+
+      - name: Run scoring harness against starter
+        run: python score.py

.gitignore

@@ -0,0 +1,7 @@
+__pycache__/
+.pytest_cache/
+.mypy_cache/
+.coverage
+.venv/
+venv/
+sharedringbuffercorrect.py

LICENSE

@@ -0,0 +1,63 @@
+# UCI Rocket Project Liquids
+
+Spring 2026 recruitment coding challenge: implement a shared ring buffer in Python.
+
+## What Students Work On
+
+Your task is to replace the intentionally bad starter in `solution.py` with a correct implementation of:
+
+- `RingSpec`
+- `SharedRingBuffer`
+
+The reference contract is defined by the official tests in `tests/official/`.
+
+## Repo Layout
+
+- `solution.py`: starter submission file. It is intentionally wrong.
+- `tests/official/`: official grading tests.
+- `tests/student/student_test_template.py`: example file you can copy to write your own tests.
+- `score.py`: simple local scoring script.
+
+## Getting Started
+
+1. Create a virtual environment.
+2. Install dependencies:
+
+```bash
+python -m pip install -r requirements.txt
+```
+
+3. Run the official score against your submission:
+
+```bash
+python score.py
+```
+
+## Student Workflow
+
+- Edit `solution.py`.
+- Leave `tests/official/` alone when you want a real score.
+- Copy `tests/student/student_test_template.py` to something like `tests/student/test_my_solution.py`.
+- Add your own cases there as you build.
+
+To run your own tests too:
+
+```bash
+python score.py --include-student-tests
+```
+
+## Challenge Contract
+
+Your implementation should preserve the public API used by the tests, including:
+
+- shared-memory backed storage via `multiprocessing.shared_memory`
+- per-reader positions and alive flags
+- wrap-around read and write memory views
+- simple byte-copy helpers
+- correct cleanup and unlink behavior
+
+If you change signatures or remove attributes the official tests will fail.
+
+## GitHub
+
+This repo includes a basic GitHub Actions workflow that smoke-tests the starter and scoring harness on pushes and pull requests.

README.md

@@ -0,0 +1 @@
+numpy>=1.26,<3

requirements.txt

@@ -0,0 +1,109 @@
+from __future__ import annotations
+
+import argparse
+from dataclasses import dataclass
+import io
+import os
+from pathlib import Path
+import sys
+import unittest
+
+
+ROOT = Path(__file__).resolve().parent
+
+
+@dataclass
+class SuiteSummary:
+    label: str
+    total: int
+    passed: int
+    failures: int
+    errors: int
+    skipped: int
+
+    @property
+    def successful(self) -> bool:
+        return self.failures == 0 and self.errors == 0
+
+
+def discover(relative_path: str) -> unittest.TestSuite:
+    loader = unittest.TestLoader()
+    return loader.discover(
+        start_dir=str(ROOT / relative_path),
+        pattern="test_*.py",
+        top_level_dir=str(ROOT),
+    )
+
+
+def summarize(label: str, result: unittest.TestResult) -> SuiteSummary:
+    failures = len(result.failures) + len(getattr(result, "unexpectedSuccesses", []))
+    errors = len(result.errors)
+    skipped = len(getattr(result, "skipped", [])) + len(getattr(result, "expectedFailures", []))
+    passed = max(0, result.testsRun - failures - errors - skipped)
+    return SuiteSummary(
+        label=label,
+        total=result.testsRun,
+        passed=passed,
+        failures=failures,
+        errors=errors,
+        skipped=skipped,
+    )
+
+
+def run_suite(label: str, relative_path: str, *, verbosity: int) -> SuiteSummary:
+    print(f"\n== {label} ==")
+    suite = discover(relative_path)
+    stream = sys.stdout if verbosity > 1 else io.StringIO()
+    runner = unittest.TextTestRunner(stream=stream, verbosity=verbosity)
+    result = runner.run(suite)
+    summary = summarize(label, result)
+    print(
+        f"{summary.label}: {summary.passed}/{summary.total} passed "
+        f"({summary.failures} failures, {summary.errors} errors, {summary.skipped} skipped)"
+    )
+    if verbosity == 1 and not summary.successful:
+        print("Run with --verbose to see full failure details.")
+    return summary
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="Run the shared ring buffer score harness.")
+    parser.add_argument("--module", default="solution", help="Module name to score. Defaults to solution.")
+    parser.add_argument(
+        "--include-student-tests",
+        action="store_true",
+        help="Also run tests under tests/student.",
+    )
+    parser.add_argument(
+        "--strict",
+        action="store_true",
+        help="Exit non-zero when any official test fails.",
+    )
+    parser.add_argument("--verbose", action="store_true", help="Use verbose unittest output.")
+    args = parser.parse_args()
+
+    os.environ["RING_BUFFER_MODULE"] = args.module
+    verbosity = 2 if args.verbose else 1
+
+    print(f"Scoring module: {args.module}")
+    official = run_suite("Official Tests", "tests/official", verbosity=verbosity)
+
+    student = None
+    if args.include_student_tests:
+        student = run_suite("Student Tests", "tests/student", verbosity=verbosity)
+
+    if student is None:
+        print(f"\nCurrent official score: {official.passed}/{official.total}")
+    else:
+        print(
+            f"\nCurrent score: {official.passed}/{official.total} official, "
+            f"{student.passed}/{student.total} student"
+        )
+
+    if args.strict and not official.successful:
+        return 1
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

score.py

@@ -0,0 +1,127 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import TypeAlias
+
+
+RingView: TypeAlias = tuple[memoryview, memoryview | None, int, bool]
+
+__all__ = ["RingSpec", "SharedRingBuffer"]
+
+
+def _todo(method: str) -> NotImplementedError:
+    return NotImplementedError(
+        f"{method} is still a TODO. Right now this file is a ring buffer in the same way "
+        f"a cardboard tube is a liquid engine."
+    )
+
+
+@dataclass(frozen=True)
+class RingSpec:
+    name: str
+    size: int
+    num_readers: int
+    reader: int = -1
+    cache_align: bool = False
+    cache_size: int = 64
+
+    def __post_init__(self) -> None:
+        if self.size <= 0:
+            raise ValueError("size must be > 0")
+        if self.num_readers < 1:
+            raise ValueError("num_readers must be >= 1")
+        if self.cache_align:
+            if self.cache_size <= 0:
+                raise ValueError("cache_size must be > 0 when cache_align is True")
+            if self.cache_size & (self.cache_size - 1):
+                raise ValueError("cache_size must be a power of two when cache_align is True")
+
+    def to_kwargs(self, *, create: bool, reader: int) -> dict:
+        return {
+            "name": self.name,
+            "create": create,
+            "size": self.size,
+            "num_readers": self.num_readers,
+            "reader": reader,
+            "cache_align": self.cache_align,
+            "cache_size": self.cache_size,
+        }
+
+
+class SharedRingBuffer:
+    _NO_READER = -1
+
+    def __init__(
+        self,
+        name: str,
+        create: bool,
+        size: int,
+        num_readers: int,
+        reader: int,
+        cache_align: bool = False,
+        cache_size: int = 64,
+    ):
+        raise _todo("__init__")
+
+    def close(self) -> None:
+        raise _todo("close")
+
+    def unlink(self) -> None:
+        raise _todo("unlink")
+
+    def __enter__(self) -> "SharedRingBuffer":
+        raise _todo("__enter__")
+
+    def __exit__(self, *_):
+        raise _todo("__exit__")
+
+    def calculate_pressure(self) -> int:
+        raise _todo("calculate_pressure")
+
+    def int_to_pos(self, value: int) -> int:
+        raise _todo("int_to_pos")
+
+    def update_reader_pos(self, new_reader_pos: int) -> None:
+        raise _todo("update_reader_pos")
+
+    def set_reader_active(self, active: bool) -> None:
+        raise _todo("set_reader_active")
+
+    def is_reader_active(self) -> bool:
+        raise _todo("is_reader_active")
+
+    def update_write_pos(self, new_writer_pos: int) -> None:
+        raise _todo("update_write_pos")
+
+    def inc_writer_pos(self, inc_amount: int) -> None:
+        raise _todo("inc_writer_pos")
+
+    def inc_reader_pos(self, inc_amount: int) -> None:
+        raise _todo("inc_reader_pos")
+
+    def get_write_pos(self):
+        raise _todo("get_write_pos")
+
+    def compute_max_amount_writable(self, force_rescan: bool = False) -> int:
+        raise _todo("compute_max_amount_writable")
+
+    def jump_to_writer(self) -> None:
+        raise _todo("jump_to_writer")
+
+    def expose_writer_mem_view(self, size: int) -> RingView:
+        raise _todo("expose_writer_mem_view")
+
+    def expose_reader_mem_view(self, size: int) -> RingView:
+        raise _todo("expose_reader_mem_view")
+
+    def simple_write(self, writer_mem_view: RingView, src: object) -> None:
+        raise _todo("simple_write")
+
+    def simple_read(self, reader_mem_view: RingView, dst: object) -> None:
+        raise _todo("simple_read")
+
+    def write_array(self, arr) -> int:
+        raise _todo("write_array")
+
+    def read_array(self, nbytes: int, dtype):
+        raise _todo("read_array")

solution.py

@@ -0,0 +1 @@
+# Test package for the shared ring buffer challenge.

tests/__init__.py

@@ -0,0 +1 @@
+# Official scoring tests.