Removed unreachable defensive checks and added tests for edge case display states.
This commit is contained in:
+11
-37
@@ -80,39 +80,23 @@ class CommandHandler:
|
|||||||
# Try to add a new subscription for this stream in this room
|
# Try to add a new subscription for this stream in this room
|
||||||
try:
|
try:
|
||||||
await self.subscription_repo.add(stream_domain, evt.room_id)
|
await self.subscription_repo.add(stream_domain, evt.room_id)
|
||||||
except sqlite3.IntegrityError as exception:
|
except sqlite3.IntegrityError:
|
||||||
# Was it a duplicate row?
|
# Room is already subscribed.
|
||||||
if "UNIQUE constraint failed" in exception.args[0]:
|
await evt.reply(
|
||||||
# Expected: room is already subscribed.
|
"This room is already subscribed to notifications for "
|
||||||
await evt.reply(
|
+ stream_domain
|
||||||
"This room is already subscribed to notifications for "
|
+ "."
|
||||||
+ stream_domain
|
|
||||||
+ "."
|
|
||||||
)
|
|
||||||
return
|
|
||||||
# Something unexpected happened. Give up.
|
|
||||||
self.log.error(
|
|
||||||
f"[{stream_domain}] An error occurred while "
|
|
||||||
f"attempting to add subscription in room "
|
|
||||||
f"{evt.room_id}: {exception}"
|
|
||||||
)
|
)
|
||||||
raise exception
|
return
|
||||||
|
|
||||||
# Try to add a placeholder row for the stream's state.
|
# Try to add a placeholder row for the stream's state.
|
||||||
try:
|
try:
|
||||||
await self.stream_repo.create(stream_domain)
|
await self.stream_repo.create(stream_domain)
|
||||||
# First time seeing this stream. Log it.
|
# First time seeing this stream. Log it.
|
||||||
self.log.info(f"[{stream_domain}] Discovered new stream!")
|
self.log.info(f"[{stream_domain}] Discovered new stream!")
|
||||||
except sqlite3.IntegrityError as exception:
|
except sqlite3.IntegrityError:
|
||||||
# Adding rows for known streams is expected.
|
# Adding rows for known streams is expected.
|
||||||
if "UNIQUE constraint failed" not in exception.args[0]:
|
pass
|
||||||
# Something unexpected happened. Give up.
|
|
||||||
self.log.error(
|
|
||||||
f"[{stream_domain}] An error occurred while "
|
|
||||||
f"attempting to add stream information "
|
|
||||||
f"after adding subscription: {exception}"
|
|
||||||
)
|
|
||||||
raise exception
|
|
||||||
|
|
||||||
# All went well! Tell the user.
|
# All went well! Tell the user.
|
||||||
self.log.info(f"[{stream_domain}] Subscription added for room {evt.room_id}.")
|
self.log.info(f"[{stream_domain}] Subscription added for room {evt.room_id}.")
|
||||||
@@ -205,13 +189,6 @@ class CommandHandler:
|
|||||||
# Get the stream state from the database
|
# Get the stream state from the database
|
||||||
stream_state = await self.stream_repo.get_by_domain(domain)
|
stream_state = await self.stream_repo.get_by_domain(domain)
|
||||||
|
|
||||||
if stream_state is None:
|
|
||||||
# Stream in subscriptions but not streams table
|
|
||||||
body_text += f"- **{domain}** \n"
|
|
||||||
body_text += " - Status: Unknown \n"
|
|
||||||
body_text += f" - Link: https://{domain}\n\n"
|
|
||||||
continue
|
|
||||||
|
|
||||||
# Determine stream name (use domain as fallback)
|
# Determine stream name (use domain as fallback)
|
||||||
stream_name = stream_state.name if stream_state.name else domain
|
stream_name = stream_state.name if stream_state.name else domain
|
||||||
safe_stream_name = sanitize_for_markdown(stream_name)
|
safe_stream_name = sanitize_for_markdown(stream_name)
|
||||||
@@ -227,11 +204,8 @@ class CommandHandler:
|
|||||||
# Determine status and duration (as a sub-bullet)
|
# Determine status and duration (as a sub-bullet)
|
||||||
if stream_state.status == StreamStatus.ONLINE:
|
if stream_state.status == StreamStatus.ONLINE:
|
||||||
# Stream is online - use last_connect_time
|
# Stream is online - use last_connect_time
|
||||||
if stream_state.last_connect_time:
|
duration = self._format_duration(stream_state.last_connect_time)
|
||||||
duration = self._format_duration(stream_state.last_connect_time)
|
body_text += f" - Status: Online for {duration} \n"
|
||||||
body_text += f" - Status: Online for {duration} \n"
|
|
||||||
else:
|
|
||||||
body_text += " - Status: Online \n"
|
|
||||||
elif stream_state.status == StreamStatus.UNKNOWN:
|
elif stream_state.status == StreamStatus.UNKNOWN:
|
||||||
# Stream status is unknown - instance unreachable
|
# Stream status is unknown - instance unreachable
|
||||||
body_text += " - Status: Unknown (instance unreachable) \n"
|
body_text += " - Status: Unknown (instance unreachable) \n"
|
||||||
|
|||||||
@@ -127,11 +127,6 @@ class StreamMonitor:
|
|||||||
# Backoff is expected behavior, not a failure
|
# Backoff is expected behavior, not a failure
|
||||||
return True
|
return True
|
||||||
|
|
||||||
# Defensive check: old_state should always exist here since
|
|
||||||
# the stream is in the DB
|
|
||||||
if old_state is None:
|
|
||||||
return False
|
|
||||||
|
|
||||||
# Flag: first state update of a brand-new stream to avoid
|
# Flag: first state update of a brand-new stream to avoid
|
||||||
# sending notifications if it's already live.
|
# sending notifications if it's already live.
|
||||||
first_update = False
|
first_update = False
|
||||||
@@ -281,10 +276,7 @@ class StreamMonitor:
|
|||||||
# Yep. This stream is now offline. Log it.
|
# Yep. This stream is now offline. Log it.
|
||||||
update_database = True
|
update_database = True
|
||||||
self.offline_timer_cache[domain] = time.time()
|
self.offline_timer_cache[domain] = time.time()
|
||||||
if first_update:
|
self.log.info(f"[{domain}] Stream is now offline.")
|
||||||
self.log.info(f"[{domain}] Stream is offline.")
|
|
||||||
else:
|
|
||||||
self.log.info(f"[{domain}] Stream is now offline.")
|
|
||||||
|
|
||||||
# Update the database with current stream state, if needed.
|
# Update the database with current stream state, if needed.
|
||||||
if update_database:
|
if update_database:
|
||||||
|
|||||||
+54
-1
@@ -27,7 +27,7 @@ from aioresponses import aioresponses
|
|||||||
|
|
||||||
from owncastsentry.commands import CommandHandler
|
from owncastsentry.commands import CommandHandler
|
||||||
from owncastsentry.models import StreamState
|
from owncastsentry.models import StreamState
|
||||||
from owncastsentry.utils import OWNCAST_STATUS_PATH
|
from owncastsentry.utils import OWNCAST_STATUS_PATH, UNKNOWN_STATUS_THRESHOLD
|
||||||
from tests.conftest import VALID_STATUS_RESPONSE
|
from tests.conftest import VALID_STATUS_RESPONSE
|
||||||
|
|
||||||
|
|
||||||
@@ -271,6 +271,59 @@ class TestSubscriptionsCommand:
|
|||||||
"instances, use `!unsubscribe <domain>`"
|
"instances, use `!unsubscribe <domain>`"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
async def test_shows_offline_stream_without_disconnect_time(
|
||||||
|
self, maubot_test_bot, maubot_plugin
|
||||||
|
) -> None:
|
||||||
|
"""Show offline status without duration before first poll completes."""
|
||||||
|
status_url = f"https://stream.logal.dev{OWNCAST_STATUS_PATH}"
|
||||||
|
with aioresponses() as mocked:
|
||||||
|
mocked.get(
|
||||||
|
status_url,
|
||||||
|
body=json.dumps(VALID_STATUS_RESPONSE).encode(),
|
||||||
|
)
|
||||||
|
await maubot_test_bot.send("!subscribe stream.logal.dev")
|
||||||
|
|
||||||
|
# Stream row exists with no state yet - query subscriptions immediately
|
||||||
|
await maubot_test_bot.send("!subscriptions")
|
||||||
|
|
||||||
|
assert len(maubot_test_bot.responded) == 2
|
||||||
|
assert maubot_test_bot.responded[1].content.body == (
|
||||||
|
"**Subscriptions for this room (1):**\n\n"
|
||||||
|
"● **stream.logal.dev** \n"
|
||||||
|
" ○ Status: Offline\n"
|
||||||
|
" ○ Link: https://stream.logal.dev\n"
|
||||||
|
"To unsubscribe from any of these Owncast "
|
||||||
|
"instances, use `!unsubscribe <domain>`"
|
||||||
|
)
|
||||||
|
|
||||||
|
async def test_shows_unknown_stream(self, maubot_test_bot, maubot_plugin) -> None:
|
||||||
|
"""Show unknown status when instance has been unreachable."""
|
||||||
|
status_url = f"https://stream.logal.dev{OWNCAST_STATUS_PATH}"
|
||||||
|
with aioresponses() as mocked:
|
||||||
|
mocked.get(
|
||||||
|
status_url,
|
||||||
|
body=json.dumps(VALID_STATUS_RESPONSE).encode(),
|
||||||
|
)
|
||||||
|
await maubot_test_bot.send("!subscribe stream.logal.dev")
|
||||||
|
|
||||||
|
# Increment failure counter past the unknown threshold
|
||||||
|
for _ in range(UNKNOWN_STATUS_THRESHOLD + 1):
|
||||||
|
await maubot_plugin.stream_repo.increment_failure_counter(
|
||||||
|
"stream.logal.dev"
|
||||||
|
)
|
||||||
|
|
||||||
|
await maubot_test_bot.send("!subscriptions")
|
||||||
|
|
||||||
|
assert len(maubot_test_bot.responded) == 2
|
||||||
|
assert maubot_test_bot.responded[1].content.body == (
|
||||||
|
"**Subscriptions for this room (1):**\n\n"
|
||||||
|
"● **stream.logal.dev** \n"
|
||||||
|
" ○ Status: Unknown (instance unreachable)\n"
|
||||||
|
" ○ Link: https://stream.logal.dev\n"
|
||||||
|
"To unsubscribe from any of these Owncast "
|
||||||
|
"instances, use `!unsubscribe <domain>`"
|
||||||
|
)
|
||||||
|
|
||||||
@time_machine.travel(datetime(2026, 3, 13, 12, 0, 0, tzinfo=UTC))
|
@time_machine.travel(datetime(2026, 3, 13, 12, 0, 0, tzinfo=UTC))
|
||||||
async def test_shows_multiple_subscriptions(
|
async def test_shows_multiple_subscriptions(
|
||||||
self, maubot_test_bot, maubot_plugin
|
self, maubot_test_bot, maubot_plugin
|
||||||
|
|||||||
Reference in New Issue
Block a user