feat(01-02): add Peewee models, database init, and driver storage
- imptune/db/database.py: deferred SqliteDatabase with WAL + foreign_keys pragmas and init_db() - imptune/db/models.py: full schema (Client, Driver, Printer, Icon) for phases 1-5 - imptune/storage/driver_store.py: SHA256 content-addressed DriverStore with dedup - tests/test_db.py: 7 TDD tests covering all db and storage behaviors
This commit is contained in:
@@ -0,0 +1 @@
|
||||
# Database package
|
||||
@@ -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)
|
||||
@@ -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"
|
||||
@@ -0,0 +1 @@
|
||||
# Storage package
|
||||
@@ -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()
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user