DeCarabas/cry
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

More database tests, test redirect

Browse source
This commit is contained in:
John Doty 2024-07-17 09:43:25 -07:00
parent 52c12785c8
commit a08257ec76
3 changed files with 258 additions and 23 deletions
Download patch file Download diff file Expand all files Collapse all files

6
cry/cli.py
View file

@ -96,7 +96,7 @@ def subscribe(url, literal):
result = d
# Check to see if this URL is already in the database.
existing = db.load_feed(result.meta.url)
existing = db.load_feed(result.meta.url) # TODO: Replace with 'load_meta'?
if existing is not None:
click.echo(f"This feed already exists (as {result.meta.url})")
return 1
@ -129,7 +129,7 @@ def import_opml(opml_file):
click.echo(f"{url} does not seem to be a feed, skipping...")
continue
existing = db.load_feed(meta.url)
existing = db.load_feed(meta.url) # TODO: Replace with 'load_meta'?
if existing is not None:
LOG.info(f"{url} already exists (as {meta.url})")
continue
@ -151,7 +151,7 @@ def refresh(url):
db = database.Database.local()
if url:
f = db.load_feed(url)
f = db.load_feed(url) # TODO: Replace with 'load_meta'?
if f is None:
click.echo(f"Not subscribed to {url}")
return 1

39
cry/database.py
View file

@ -91,7 +91,15 @@ class Database:
if not isinstance(path, str):
path.parent.mkdir(parents=True, exist_ok=True)
db = sqlite3.Connection(str(path), autocommit=False)
db.execute("PRAGMA foreign_keys = ON")
# This is WILD that I need to do this because you cannot enable or
# disable foreign keys with a transaction active, and
# `autocommit=False` means that a transaction is *always* active.
db.executescript("COMMIT;PRAGMA foreign_keys = ON;BEGIN;")
cursor = db.execute("PRAGMA foreign_keys")
rows = cursor.fetchall()
assert str(rows[0][0]) == "1", f"Foreign keys not enabled! {rows[0][0]}"
self.db = db
self.origin = origin
@ -433,30 +441,21 @@ class Database:
"UPDATE feeds SET url = ? WHERE url = ?", [new_url, old_url]
)
else:
# Preserve the entries that were under the old url.
# First update all the entries that you can with the old url.
self.db.execute(
"""
UPDATE entries
UPDATE OR IGNORE entries
SET feed_url = ?
WHERE feed_url = ?
ON CONFLICT DO UPDATE
SET
-- NOTE: This is also part of the feed merge algorithm, BUT
-- we implement it here. See the comment in store_feed
-- for the rationale.
inserted_at=MIN(inserted_at, excluded.inserted_at),
title=CASE
WHEN inserted_at < excluded.inserted_at THEN title
ELSE excluded.title
END,
link=CASE
WHEN inserted_at < excluded.inserted_at THEN link
ELSE excluded.link
END
"""
""",
[new_url, old_url],
)
# Mark the old feed dead.
# TODO: It is expensive and not worth it to try to load and
# re-insert all the old stuff so I'm not going to
# bother.
# Mark the old feed unsubscribed.
self.db.execute(
"""
UPDATE feeds
@ -464,5 +463,5 @@ class Database:
last_fetched_ts = ?
WHERE url = ?
""",
[feed.FEED_STATUS_DEAD, int(time.time()), old_url],
[feed.FEED_STATUS_UNSUBSCRIBED, int(time.time()), old_url],
)

236
tests/test_database.py
View file

@ -1,9 +1,12 @@
import dataclasses
import pathlib
import random
import string
import tempfile
import time
from cry import database
from cry import feed
def random_slug() -> str:
@ -64,3 +67,236 @@ def test_database_prop_get_set():
val = random_slug()
db.set_property("foo", val)
assert db.get_property("foo") == val
REF_TIME = int(time.time())
FEED = feed.Feed(
meta=feed.FeedMeta(
url="http://example.com/test/feed",
last_fetched_ts=REF_TIME,
retry_after_ts=REF_TIME,
status=feed.FEED_STATUS_ALIVE,
etag=random_slug(),
modified=random_slug(),
),
title="Test Feed",
link="http://example.com/test",
entries=[
feed.Entry(
id=random_slug(),
inserted_at=(REF_TIME * 1000) + index,
title=f"Entry {index}",
link=f"http://example.com/test/a{index}",
)
for index in range(100, 0, -1)
],
)
def test_database_load_store_meta():
db = database.Database(":memory:", random_slug())
db.ensure_database_schema()
metas = db.load_all_meta()
assert metas == []
def test_database_store_feed():
db = database.Database(":memory:", random_slug())
db.ensure_database_schema()
db.store_feed(FEED)
loaded = db.load_feed(FEED.meta.url)
assert loaded == FEED
def test_database_store_feed_dups():
db = database.Database(":memory:", random_slug())
db.ensure_database_schema()
count = db.store_feed(FEED)
assert count == len(FEED.entries)
new_entries = db.store_feed(FEED)
assert new_entries == 0
def test_database_store_feed_fetch_meta():
db = database.Database(":memory:", random_slug())
db.ensure_database_schema()
db.store_feed(FEED)
meta = db.load_all_meta()
assert meta == [FEED.meta]
def test_database_store_feed_fetch_all():
db = database.Database(":memory:", random_slug())
db.ensure_database_schema()
db.store_feed(FEED)
expected = dataclasses.replace(FEED, entries=FEED.entries[:13])
all_feeds = db.load_all(feed_limit=13)
assert all_feeds == [expected]
def test_database_store_feed_fetch_all_dups():
db = database.Database(":memory:", random_slug())
db.ensure_database_schema()
db.store_feed(FEED)
db.store_feed(FEED)
all_feeds = db.load_all(feed_limit=10000)
assert all_feeds == [FEED]
def test_database_store_feed_fetch_pattern_miss():
db = database.Database(":memory:", random_slug())
db.ensure_database_schema()
db.store_feed(FEED)
expected = dataclasses.replace(FEED, entries=FEED.entries[:13])
all_feeds = db.load_all(feed_limit=13, pattern="no_existo")
assert all_feeds == []
def test_database_store_feed_fetch_pattern_url():
db = database.Database(":memory:", random_slug())
db.ensure_database_schema()
db.store_feed(FEED)
expected = dataclasses.replace(FEED, entries=FEED.entries[:13])
all_feeds = db.load_all(feed_limit=13, pattern=FEED.link)
assert all_feeds == [expected]
def test_database_store_feed_fetch_pattern_name():
db = database.Database(":memory:", random_slug())
db.ensure_database_schema()
db.store_feed(FEED)
expected = dataclasses.replace(FEED, entries=FEED.entries[:13])
all_feeds = db.load_all(feed_limit=13, pattern=FEED.title)
assert all_feeds == [expected]
def test_database_store_with_update():
db = database.Database(":memory:", random_slug())
db.ensure_database_schema()
db.store_feed(FEED)
updated_feed = dataclasses.replace(
FEED,
meta=dataclasses.replace(
FEED.meta,
last_fetched_ts=FEED.meta.last_fetched_ts + 10,
retry_after_ts=FEED.meta.retry_after_ts + 20,
# status=feed.FEED_STATUS_UNSUBSCRIBED,
etag=None,
modified=random_slug(),
),
title=FEED.title + " (updated)",
link=FEED.link + "/updated",
)
db.store_feed(updated_feed)
all_feeds = db.load_all(feed_limit=100)
assert all_feeds == [updated_feed]
def test_database_store_with_older_entries():
db = database.Database(":memory:", random_slug())
db.ensure_database_schema()
db.store_feed(FEED)
old_entry = FEED.entries[0]
older_entry = dataclasses.replace(
old_entry,
inserted_at=old_entry.inserted_at - 10,
title=old_entry.title + " (older)",
link=old_entry.link + "/older",
)
updated_feed = dataclasses.replace(FEED, entries=[older_entry])
db.store_feed(updated_feed)
all_feeds = db.load_all(feed_limit=100)
found_entries = list(
filter(
lambda e: e.id == older_entry.id,
all_feeds[0].entries,
)
)
assert found_entries == [older_entry]
def test_database_store_update_meta():
db = database.Database(":memory:", random_slug())
db.ensure_database_schema()
db.store_feed(FEED)
new_meta = dataclasses.replace(
FEED.meta,
last_fetched_ts=FEED.meta.last_fetched_ts + 10,
retry_after_ts=FEED.meta.last_fetched_ts + 20,
status=feed.FEED_STATUS_DEAD,
etag=random_slug(),
modified=random_slug(),
)
db.update_meta(new_meta)
assert db.load_all_meta()[0] == new_meta
def test_database_set_feed_status():
db = database.Database(":memory:", random_slug())
db.ensure_database_schema()
db.store_feed(FEED)
assert db.load_all_meta()[0].status != feed.FEED_STATUS_UNSUBSCRIBED
db.set_feed_status(FEED.meta.url, feed.FEED_STATUS_UNSUBSCRIBED)
# TODO: Ensure that the updated time is touched too.
assert db.load_all_meta()[0].status == feed.FEED_STATUS_UNSUBSCRIBED
def test_database_redirect_clean():
db = database.Database(":memory:", random_slug())
db.ensure_database_schema()
db.store_feed(FEED)
new_url = f"http://example.com/redirect/{random_slug()}"
db.redirect_feed(FEED.meta.url, new_url)
expected_meta = dataclasses.replace(FEED.meta, url=new_url)
assert db.load_all_meta() == [expected_meta]
expected_feed = dataclasses.replace(FEED, meta=expected_meta)
assert db.load_all(feed_limit=9999) == [expected_feed]
def test_database_redirect_with_merge():
db = database.Database(":memory:", random_slug())
db.ensure_database_schema()
db.store_feed(FEED)
new_url = f"http://example.com/redirect/{random_slug()}"
expected_meta = dataclasses.replace(FEED.meta, url=new_url)
expected_feed = dataclasses.replace(FEED, meta=expected_meta)
db.store_feed(expected_feed)
# NOTE: This is flaky because the time might shift on me.
db.redirect_feed(FEED.meta.url, new_url)
old_dead_meta = dataclasses.replace(FEED.meta, status=feed.FEED_STATUS_UNSUBSCRIBED)
assert db.load_all_meta() == [old_dead_meta, expected_meta]
assert db.load_all(feed_limit=9999) == [expected_feed]