Initial commit: GitHub Release Monitor CLI tool
CLI tool that tracks GitHub releases or tags from user-defined repositories and exposes them as a subscribable RSS 2.0 feed. Features: - Track releases or tags per repository - SQLite storage with automatic deduplication - RSS 2.0 feed with markdown-stripped release notes - Built-in HTTP server for local feed access - Rate-limit-aware GitHub API client - 64 passing tests
This commit is contained in:
@@ -0,0 +1,143 @@
|
||||
"""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 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
|
||||
Reference in New Issue
Block a user