diff --git a/imptune/db/__init__.py b/imptune/db/__init__.py new file mode 100644 index 0000000..99ce574 --- /dev/null +++ b/imptune/db/__init__.py @@ -0,0 +1 @@ +# Database package diff --git a/imptune/db/database.py b/imptune/db/database.py new file mode 100644 index 0000000..999979a --- /dev/null +++ b/imptune/db/database.py @@ -0,0 +1,29 @@ +"""Peewee SQLite database instance and initialization.""" +from peewee import SqliteDatabase + +from imptune.config import DB_PATH + +# Deferred init — path is set at runtime via init_db() so tests can override DB_PATH +db = SqliteDatabase(None) + + +def init_db() -> None: + """Initialize the SQLite database with WAL mode and foreign keys. + + Idempotent — safe to call on every application startup. + Creates all ORM tables if they do not already exist. + """ + from imptune.db.models import Client, Driver, Printer, Icon + + # Re-read DB_PATH each time so tests can patch imptune.config.DB_PATH + import imptune.config as cfg + + db.init( + cfg.DB_PATH, + pragmas={ + "journal_mode": "wal", + "foreign_keys": 1, + }, + ) + db.connect(reuse_if_open=True) + db.create_tables([Client, Driver, Printer, Icon], safe=True) diff --git a/imptune/db/models.py b/imptune/db/models.py new file mode 100644 index 0000000..bfa1551 --- /dev/null +++ b/imptune/db/models.py @@ -0,0 +1,78 @@ +"""Peewee ORM models — full schema for phases 1-5.""" +from datetime import datetime + +from peewee import ( + BooleanField, + CharField, + DateTimeField, + ForeignKeyField, + IntegerField, + Model, +) + +from imptune.db.database import db + + +class BaseModel(Model): + """Base model that binds all models to the shared db instance.""" + + class Meta: + database = db + + +class Client(BaseModel): + """Represents a deployment target (AD client / OU).""" + + name = CharField(unique=True) + created_at = DateTimeField(default=datetime.utcnow) + + class Meta: + table_name = "client" + + +class Driver(BaseModel): + """Uploaded printer driver package (content-addressed by SHA256).""" + + sha256 = CharField(unique=True, index=True) + original_filename = CharField() + size_bytes = IntegerField() + uploaded_at = DateTimeField(default=datetime.utcnow) + driver_desc = CharField(null=True) + inf_filename = CharField(null=True) + architecture = CharField(null=True) + has_cat_file = BooleanField(default=False) + + class Meta: + table_name = "driver" + + +class Printer(BaseModel): + """Printer configuration record.""" + + name = CharField() + ip_address = CharField() + port_name = CharField() + client = ForeignKeyField(Client, null=True, backref="printers") + driver = ForeignKeyField(Driver, null=True, backref="printers") + duplex_mode = CharField(default="OneSided") + color_mode = BooleanField(default=True) + paper_size = CharField(default="A4") + collate = BooleanField(default=True) + created_at = DateTimeField(default=datetime.utcnow) + updated_at = DateTimeField(default=datetime.utcnow) + + class Meta: + table_name = "printer" + + +class Icon(BaseModel): + """Printer icon image (one per printer).""" + + printer = ForeignKeyField(Printer, unique=True, backref="icons") + sha256 = CharField() + original_filename = CharField() + size_bytes = IntegerField() + uploaded_at = DateTimeField(default=datetime.utcnow) + + class Meta: + table_name = "icon" diff --git a/imptune/storage/__init__.py b/imptune/storage/__init__.py new file mode 100644 index 0000000..69a52f6 --- /dev/null +++ b/imptune/storage/__init__.py @@ -0,0 +1 @@ +# Storage package diff --git a/imptune/storage/driver_store.py b/imptune/storage/driver_store.py new file mode 100644 index 0000000..0b2e4c7 --- /dev/null +++ b/imptune/storage/driver_store.py @@ -0,0 +1,34 @@ +"""Content-addressed file storage for driver packages.""" +import hashlib +from pathlib import Path + + +class DriverStore: + """SHA256 content-addressed file storage for printer driver packages. + + Files are stored as DRIVERS_DIR/{sha256} so identical uploads are + deduplicated automatically. + """ + + def __init__(self, base_dir: str) -> None: + self._base = Path(base_dir) + self._base.mkdir(parents=True, exist_ok=True) + + def save(self, data: bytes) -> str: + """Persist *data* and return its SHA256 hex digest. + + If the file already exists the write is skipped (deduplication). + """ + digest = hashlib.sha256(data).hexdigest() + dest = self._base / digest + if not dest.exists(): + dest.write_bytes(data) + return digest + + def get_path(self, sha256: str) -> Path: + """Return the filesystem path for a given SHA256 digest.""" + return self._base / sha256 + + def exists(self, sha256: str) -> bool: + """Return True if the file for *sha256* exists on disk.""" + return (self._base / sha256).exists() diff --git a/tests/test_db.py b/tests/test_db.py new file mode 100644 index 0000000..01ebeb7 --- /dev/null +++ b/tests/test_db.py @@ -0,0 +1,141 @@ +"""Tests for database initialization and driver storage.""" +import hashlib +from pathlib import Path + +import pytest + + +@pytest.fixture +def db_env(tmp_path, monkeypatch): + """Set up temp DATA_DIR and configure db to use temp paths.""" + data_dir = tmp_path / "data" + data_dir.mkdir() + drivers_dir = data_dir / "drivers" + drivers_dir.mkdir() + db_path = str(data_dir / "imptune.db") + + monkeypatch.setenv("DATA_DIR", str(data_dir)) + + import imptune.config as cfg + cfg.DATA_DIR = str(data_dir) + cfg.DB_PATH = db_path + cfg.DRIVERS_DIR = str(drivers_dir) + + # Close and re-init the db with the temp path + from imptune.db.database import db + if not db.is_closed(): + db.close() + + return { + "data_dir": data_dir, + "drivers_dir": drivers_dir, + "db_path": db_path, + } + + +def test_create_tables(db_env): + """init_db() creates all 4 tables in a fresh SQLite file.""" + from imptune.db.database import db, init_db + + if not db.is_closed(): + db.close() + + init_db() + + tables = db.get_tables() + assert "client" in tables + assert "driver" in tables + assert "printer" in tables + assert "icon" in tables + + db.close() + + +def test_wal_mode(db_env): + """After init_db(), PRAGMA journal_mode returns 'wal'.""" + from imptune.db.database import db, init_db + + if not db.is_closed(): + db.close() + + init_db() + + cursor = db.execute_sql("PRAGMA journal_mode;") + mode = cursor.fetchone()[0] + assert mode == "wal" + + db.close() + + +def test_foreign_keys(db_env): + """After init_db(), PRAGMA foreign_keys returns 1.""" + from imptune.db.database import db, init_db + + if not db.is_closed(): + db.close() + + init_db() + + cursor = db.execute_sql("PRAGMA foreign_keys;") + value = cursor.fetchone()[0] + assert value == 1 + + db.close() + + +def test_idempotent(db_env): + """Calling init_db() twice does not raise an error.""" + from imptune.db.database import db, init_db + + if not db.is_closed(): + db.close() + + init_db() + db.close() + init_db() # second call — must not raise + + db.close() + + +def test_driver_store_save(db_env, tmp_path): + """Saving bytes returns their SHA256 hex digest and creates the file.""" + from imptune.storage.driver_store import DriverStore + + drivers_dir = db_env["drivers_dir"] + store = DriverStore(str(drivers_dir)) + + data = b"test driver package content" + expected_sha256 = hashlib.sha256(data).hexdigest() + + result = store.save(data) + + assert result == expected_sha256 + assert (drivers_dir / expected_sha256).exists() + + +def test_driver_store_dedup(db_env): + """Saving the same bytes twice results in one file on disk.""" + from imptune.storage.driver_store import DriverStore + + drivers_dir = db_env["drivers_dir"] + store = DriverStore(str(drivers_dir)) + + data = b"duplicate driver data" + store.save(data) + store.save(data) + + files = list(drivers_dir.iterdir()) + assert len(files) == 1 + + +def test_driver_store_get_path(db_env): + """get_path(sha256) returns the correct file path.""" + from imptune.storage.driver_store import DriverStore + + drivers_dir = db_env["drivers_dir"] + store = DriverStore(str(drivers_dir)) + + sha256 = "abcdef1234567890" * 4 # 64 hex chars + path = store.get_path(sha256) + + assert path == Path(str(drivers_dir)) / sha256