ARIA
5cd809930e
Add image_url column to repos table with migration, CLI image add/remove commands, and Media RSS <media:content> elements in generated feed output. New CLI commands: ghrel image add <repo> <url>, ghrel image remove <repo>
219 lines
8.4 KiB
Python
219 lines
8.4 KiB
Python
"""Tests for the database layer."""
|
|
|
|
import os
|
|
import stat
|
|
import sqlite3
|
|
import tempfile
|
|
|
|
import pytest
|
|
|
|
from src import db
|
|
|
|
|
|
@pytest.fixture
|
|
def conn():
|
|
"""Provide an in-memory database connection for testing."""
|
|
c = sqlite3.connect(":memory:")
|
|
c.execute("PRAGMA foreign_keys=ON")
|
|
db._create_tables(c)
|
|
return c
|
|
|
|
|
|
class TestCreateTables:
|
|
def test_tables_exist(self, conn):
|
|
tables = set(r[0] for r in conn.execute(
|
|
"SELECT name FROM sqlite_master WHERE type='table'"
|
|
).fetchall())
|
|
assert "repos" in tables
|
|
assert "entries" in tables
|
|
|
|
def test_mode_check_constraint(self, conn):
|
|
with pytest.raises(sqlite3.IntegrityError):
|
|
conn.execute("INSERT INTO repos (owner_repo, mode) VALUES ('a/b', 'invalid')")
|
|
conn.commit()
|
|
|
|
def test_kind_check_constraint(self, conn):
|
|
conn.execute("INSERT INTO repos (owner_repo) VALUES ('a/b')")
|
|
conn.commit()
|
|
with pytest.raises(sqlite3.IntegrityError):
|
|
conn.execute(
|
|
"INSERT INTO entries (repo_id, kind, tag_name, title, published_at, html_url) "
|
|
"VALUES (1, 'invalid', 'v1', 'v1', '2025-01-01', 'http://x')"
|
|
)
|
|
conn.commit()
|
|
|
|
|
|
class TestAddRepo:
|
|
def test_add_repo_default_mode(self, conn):
|
|
repo_id = db.add_repo(conn, "owner/repo")
|
|
assert repo_id == 1
|
|
row = conn.execute("SELECT owner_repo, mode FROM repos WHERE id = ?", (repo_id,)).fetchone()
|
|
assert row == ("owner/repo", "release")
|
|
|
|
def test_add_repo_tag_mode(self, conn):
|
|
repo_id = db.add_repo(conn, "owner/repo", "tag")
|
|
row = conn.execute("SELECT mode FROM repos WHERE id = ?", (repo_id,)).fetchone()
|
|
assert row[0] == "tag"
|
|
|
|
def test_add_repo_duplicate_fails(self, conn):
|
|
db.add_repo(conn, "owner/repo")
|
|
with pytest.raises(sqlite3.IntegrityError):
|
|
db.add_repo(conn, "owner/repo")
|
|
|
|
|
|
class TestRemoveRepo:
|
|
def test_remove_existing(self, conn):
|
|
db.add_repo(conn, "owner/repo")
|
|
assert db.remove_repo(conn, "owner/repo") is True
|
|
assert db.list_repos(conn) == []
|
|
|
|
def test_remove_nonexistent(self, conn):
|
|
assert db.remove_repo(conn, "no/one") is False
|
|
|
|
def test_remove_cascades_to_entries(self, conn):
|
|
repo_id = db.add_repo(conn, "owner/repo")
|
|
db.upsert_entry(conn, repo_id, "release", "v1", "v1", "", "2025-01-01", "http://x")
|
|
db.remove_repo(conn, "owner/repo")
|
|
assert db.get_entry_count(conn, repo_id) == 0
|
|
|
|
|
|
class TestUpdateRepoMode:
|
|
def test_update_mode(self, conn):
|
|
db.add_repo(conn, "owner/repo")
|
|
assert db.update_repo_mode(conn, "owner/repo", "tag") is True
|
|
row = conn.execute("SELECT mode FROM repos WHERE owner_repo = 'owner/repo'").fetchone()
|
|
assert row[0] == "tag"
|
|
|
|
def test_update_nonexistent(self, conn):
|
|
assert db.update_repo_mode(conn, "no/one", "tag") is False
|
|
|
|
|
|
class TestListRepos:
|
|
def test_empty_list(self, conn):
|
|
assert db.list_repos(conn) == []
|
|
|
|
def test_list_with_entries(self, conn):
|
|
repo_id = db.add_repo(conn, "a/b")
|
|
db.upsert_entry(conn, repo_id, "release", "v1", "v1", "", "2025-01-01", "http://x")
|
|
result = db.list_repos(conn)
|
|
assert len(result) == 1
|
|
assert result[0]["owner_repo"] == "a/b"
|
|
assert result[0]["entries"] == 1
|
|
assert result[0]["mode"] == "release"
|
|
|
|
|
|
class TestUpsertEntry:
|
|
def test_insert_new(self, conn):
|
|
repo_id = db.add_repo(conn, "owner/repo")
|
|
assert db.upsert_entry(conn, repo_id, "release", "v1", "Release v1", "body text",
|
|
"2025-01-01T00:00:00Z", "http://example.com") is True
|
|
|
|
def test_skip_duplicate(self, conn):
|
|
repo_id = db.add_repo(conn, "owner/repo")
|
|
assert db.upsert_entry(conn, repo_id, "release", "v1", "v1", "", "2025-01-01", "http://x") is True
|
|
assert db.upsert_entry(conn, repo_id, "release", "v1", "v1", "", "2025-01-01", "http://x") is False
|
|
|
|
|
|
class TestGetAllEntries:
|
|
def test_returns_entries_sorted_desc(self, conn):
|
|
repo_id = db.add_repo(conn, "owner/repo")
|
|
db.upsert_entry(conn, repo_id, "release", "v1", "v1", "", "2025-01-01", "http://x")
|
|
db.upsert_entry(conn, repo_id, "release", "v2", "v2", "", "2025-01-02", "http://y")
|
|
entries = db.get_all_entries(conn)
|
|
assert entries[0]["tag_name"] == "v2"
|
|
assert entries[1]["tag_name"] == "v1"
|
|
|
|
def test_respects_limit(self, conn):
|
|
repo_id = db.add_repo(conn, "owner/repo")
|
|
for i in range(5):
|
|
db.upsert_entry(conn, repo_id, "release", f"v{i}", f"v{i}", "", f"2025-01-0{i}", "http://x")
|
|
entries = db.get_all_entries(conn, limit=3)
|
|
assert len(entries) == 3
|
|
|
|
|
|
class TestRepoImage:
|
|
def test_set_image(self, conn):
|
|
db.add_repo(conn, "owner/repo")
|
|
assert db.set_repo_image(conn, "owner/repo", "https://example.com/logo.png") is True
|
|
row = conn.execute("SELECT image_url FROM repos WHERE owner_repo = 'owner/repo'").fetchone()
|
|
assert row[0] == "https://example.com/logo.png"
|
|
|
|
def test_set_image_nonexistent(self, conn):
|
|
assert db.set_repo_image(conn, "no/one", "https://example.com/logo.png") is False
|
|
|
|
def test_set_image_overwrites(self, conn):
|
|
db.add_repo(conn, "owner/repo")
|
|
db.set_repo_image(conn, "owner/repo", "https://example.com/old.png")
|
|
db.set_repo_image(conn, "owner/repo", "https://example.com/new.png")
|
|
row = conn.execute("SELECT image_url FROM repos WHERE owner_repo = 'owner/repo'").fetchone()
|
|
assert row[0] == "https://example.com/new.png"
|
|
|
|
def test_remove_image(self, conn):
|
|
db.add_repo(conn, "owner/repo")
|
|
db.set_repo_image(conn, "owner/repo", "https://example.com/logo.png")
|
|
assert db.remove_repo_image(conn, "owner/repo") is True
|
|
row = conn.execute("SELECT image_url FROM repos WHERE owner_repo = 'owner/repo'").fetchone()
|
|
assert row[0] is None
|
|
|
|
def test_remove_image_no_image_set(self, conn):
|
|
db.add_repo(conn, "owner/repo")
|
|
assert db.remove_repo_image(conn, "owner/repo") is False
|
|
|
|
def test_image_url_in_entries(self, conn):
|
|
repo_id = db.add_repo(conn, "owner/repo")
|
|
db.set_repo_image(conn, "owner/repo", "https://example.com/logo.png")
|
|
db.upsert_entry(conn, repo_id, "release", "v1", "v1", "", "2025-01-01", "http://x")
|
|
entries = db.get_all_entries(conn)
|
|
assert len(entries) == 1
|
|
assert entries[0]["image_url"] == "https://example.com/logo.png"
|
|
|
|
def test_image_url_none_by_default(self, conn):
|
|
repo_id = db.add_repo(conn, "owner/repo")
|
|
db.upsert_entry(conn, repo_id, "release", "v1", "v1", "", "2025-01-01", "http://x")
|
|
entries = db.get_all_entries(conn)
|
|
assert entries[0]["image_url"] is None
|
|
|
|
def test_migration_adds_column(self):
|
|
"""Test that _migrate_add_image_url adds the column if missing."""
|
|
c = sqlite3.connect(":memory:")
|
|
c.execute("PRAGMA foreign_keys=ON")
|
|
c.execute("""
|
|
CREATE TABLE repos (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
owner_repo TEXT NOT NULL UNIQUE,
|
|
mode TEXT NOT NULL DEFAULT 'release'
|
|
)
|
|
""")
|
|
c.execute("""
|
|
CREATE TABLE entries (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
repo_id INTEGER NOT NULL,
|
|
kind TEXT NOT NULL CHECK(kind IN ('release', 'tag')),
|
|
tag_name TEXT NOT NULL,
|
|
title TEXT NOT NULL,
|
|
body TEXT DEFAULT '',
|
|
published_at DATETIME NOT NULL,
|
|
html_url TEXT NOT NULL,
|
|
fetched_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
FOREIGN KEY (repo_id) REFERENCES repos(id) ON DELETE CASCADE,
|
|
UNIQUE(repo_id, tag_name)
|
|
)
|
|
""")
|
|
c.commit()
|
|
db._migrate_add_image_url(c)
|
|
columns = [col[1] for col in c.execute("PRAGMA table_info(repos)").fetchall()]
|
|
assert "image_url" in columns
|
|
c.close()
|
|
|
|
|
|
class TestGetConnection:
|
|
def test_creates_file(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
path = os.path.join(tmp, "test.db")
|
|
c = db.get_connection(path)
|
|
c.close()
|
|
assert os.path.exists(path)
|
|
# Check permissions
|
|
mode = os.stat(path).st_mode
|
|
assert stat.S_IMODE(mode) == 0o600
|