Compare commits

..

No commits in common. "43f5a1c64cf9c902f58549e1a3dbb266a2d21a7f" and "09f04ea60bb7bc8ff64178252bc7146a9abd12b1" have entirely different histories.

9 changed files with 43 additions and 488 deletions

View file

@ -1,17 +0,0 @@
name: Tests
on:
pull_request:
push:
branches: [main]
jobs:
unit-tests:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: '3.11'
- name: Run unit tests
run: python -m unittest discover -s tests -v

View file

@ -1,29 +0,0 @@
# AGENTS.md — Argus
If you're an AI agent using Argus, read this first.
## What Argus Does
Argus tracks GitHub/Forgejo notifications, logs your community actions, and escalates things you miss. It's your eyes on the repos you care about.
## Setup
See [docs/SKILL.md](docs/SKILL.md) for installation and daily usage.
## Behavioral Guidelines
**Read [docs/behavioral-guidelines.md](docs/behavioral-guidelines.md) before interacting with any human on GitHub.**
Key rules:
- Don't engage with hostile users — dismiss and escalate to your operator
- Be concise, specific, and humble in all public comments
- Escalate to a human when unsure — a missed comment beats a bad one
- Every notification must be acted on or dismissed with a reason
## Procedures
- [Notification triage & PR review workflow](docs/procedures/subagent-completion.md)
## Contributing
If you find a bug or want to improve Argus, open a PR. Follow the repo's coding style (shell + Python, keep it simple).

View file

@ -1,4 +0,0 @@
.PHONY: test
test:
python3 -m unittest discover -s tests -v

View file

@ -1,7 +1,5 @@
# Argus 👁️
[![Tests](https://github.com/al-munazzim/argus/actions/workflows/tests.yml/badge.svg)](https://github.com/al-munazzim/argus/actions/workflows/tests.yml)
CLI tool for GitHub/Forgejo community awareness — tracking notifications, logging activity, managing escalations.
Named after **Argus Panoptes**, the all-seeing giant of Greek mythology.
@ -56,12 +54,6 @@ For AI agents using OpenClaw, add this to your `HEARTBEAT.md` for periodic commu
The install script outputs this block — just copy it to your HEARTBEAT.md.
## Development
```bash
make test
```
## Commands
### Core
@ -180,3 +172,26 @@ journalctl -u argus -f # View logs
## License
MIT
## Author
Zeus @ Olymp
## Port Convention
For multi-agent deployments, use this formula:
```
ARGUS_PORT = 8100 + (UID mod 100) * 10
```
| Agent | UID | Dashboard | Datasette |
|--------|------|-----------|-----------|
| Zeus | 0 | 8100 | 8101 |
| Doxios | 1002 | 8120 | 8121 |
| Hermes | 1003 | 8130 | 8131 |
Set in systemd service or environment:
```bash
Environment=ARGUS_PORT=8130
```

226
bin/argus
View file

@ -19,7 +19,7 @@ from typing import Optional
# Configuration
# ============================================================================
VERSION = "0.5.0"
VERSION = "0.4.0"
DEFAULT_DB_PATH = Path.home() / ".argus" / "argus.db"
DEFAULT_PORT = 8100 # Dashboard port; datasette runs on port+1
@ -590,44 +590,17 @@ def cmd_notif_pull(args) -> None:
new_count = 0
kept_count = 0
dismissed_count = 0
reopened_count = 0
for notif in notifications:
notif_id = notif["id"]
# Check if already exists
existing = conn.execute(
"SELECT id, reason, dismissed FROM notifications WHERE id = ?",
"SELECT id FROM notifications WHERE id = ?",
(notif_id,)
).fetchone()
if existing:
old_reason = existing[1]
was_dismissed = existing[2]
new_reason = notif["reason"]
# Only re-open if the reason actually changed (e.g. subscribed → mention).
# If dismissed with same reason, GitHub is just re-serving the same
# notification — mark it done on GitHub to stop the loop.
if old_reason != new_reason:
conn.execute(
"UPDATE notifications SET reason = ?, dismissed = 0, "
"action_taken = NULL, acted_at = NULL, "
"filter_reason = 'RE-OPENED: new activity detected (was dismissed, reason: ' || ? || ' → ' || ? || ')' "
"WHERE id = ?",
(new_reason, old_reason, new_reason, notif_id)
)
conn.commit()
reopened_count += 1
elif was_dismissed and not args.dry_run and backend == "gh":
# Mark as done on GitHub to prevent re-open loop
try:
subprocess.run(
["gh", "api", "--method", "DELETE",
f"notifications/threads/{notif_id}"],
capture_output=True, text=True, check=False
)
except Exception:
pass
continue
# Match against pre-fetched own PRs (O(1) lookup, no API call)
@ -678,17 +651,6 @@ def cmd_notif_pull(args) -> None:
dismissed_count += 1
else:
kept_count += 1
# Mark as done on GitHub (DELETE removes from inbox entirely)
if not args.dry_run and backend == "gh":
try:
subprocess.run(
["gh", "api", "--method", "DELETE",
f"notifications/threads/{notif_id}"],
capture_output=True, text=True, check=False
)
except Exception:
pass # Best-effort — don't fail the pull if marking done fails
if not args.dry_run:
# Record poll
@ -701,8 +663,7 @@ def cmd_notif_pull(args) -> None:
conn.close()
prefix = "[DRY-RUN] " if args.dry_run else "✓ "
reopen_msg = f", {reopened_count} re-opened" if reopened_count else ""
print(f"\n{prefix}Pulled {new_count} new ({kept_count} kept, {dismissed_count} auto-dismissed{reopen_msg})")
print(f"\n{prefix}Pulled {new_count} new ({kept_count} kept, {dismissed_count} auto-dismissed)")
def _migrate_filter_columns(conn: sqlite3.Connection) -> None:
@ -715,8 +676,6 @@ def _migrate_filter_columns(conn: sqlite3.Connection) -> None:
"filter_reason": "TEXT",
"is_own_pr": "INTEGER DEFAULT 0",
"ci_status": "TEXT",
"processing": "INTEGER DEFAULT 0",
"tags": "TEXT DEFAULT ''",
}
for col, typedef in migrations.items():
if col not in columns:
@ -746,7 +705,7 @@ def cmd_notif_list(args) -> None:
params = []
if args.pending:
conditions.append("dismissed = 0 AND acted_at IS NULL AND processing = 0")
conditions.append("dismissed = 0 AND acted_at IS NULL")
if getattr(args, 'dismissed', False):
conditions.append("dismissed = 1")
if getattr(args, 'own', False):
@ -767,18 +726,10 @@ def cmd_notif_list(args) -> None:
return
for n in notifications:
try:
locked = n["processing"]
except (IndexError, KeyError):
locked = 0
status = "✓" if n["acted_at"] else ("✗" if n["dismissed"] else ("🔒" if locked else "○"))
status = "✓" if n["acted_at"] else ("✗" if n["dismissed"] else "○")
own_tag = " 🔵" if n["is_own_pr"] else ""
ci_tag = f" [CI:{n['ci_status']}]" if n["ci_status"] else ""
try:
tags_str = f" 🏷{n['tags']}" if n["tags"] else ""
except (IndexError, KeyError):
tags_str = ""
print(f"{status}{own_tag}{ci_tag}{tags_str} [{n['id'][:8]}] {n['title'][:60]}")
print(f"{status}{own_tag}{ci_tag} [{n['id'][:8]}] {n['title'][:60]}")
print(f" Repo: {n['repo_id']} | Reason: {n['reason']}")
if n["filter_reason"]:
print(f" Filter: {n['filter_reason']}")
@ -812,135 +763,16 @@ def cmd_notif_act(args) -> None:
""", (datetime.now().isoformat(), f"{action}: {detail}".strip(": "), notif["id"]))
# Log the activity
# sqlite3.Row doesn't have .get() — use dict() or try/except
notif_dict = dict(notif)
conn.execute("""
INSERT INTO activity_log (issue_id, action, detail, github_url)
VALUES (?, ?, ?, ?)
""", (notif_dict.get("issue_id"), action, detail or notif_dict.get("title", ""), notif_dict.get("url")))
""", (notif.get("issue_id"), action, detail or notif["title"], notif.get("url")))
conn.commit()
conn.close()
print(f"✓ Marked notification as acted: {action}")
def cmd_notif_lock(args) -> None:
"""Lock notification to prevent duplicate processing."""
conn = get_db()
notif = conn.execute(
"SELECT * FROM notifications WHERE id LIKE ?",
(f"{args.id}%",)
).fetchone()
if not notif:
print(f"Error: Notification not found: {args.id}", file=sys.stderr)
sys.exit(1)
if notif["processing"]:
print(f"⚠ Already locked: {notif['id'][:8]}")
sys.exit(1)
conn.execute("UPDATE notifications SET processing = 1 WHERE id = ?", (notif["id"],))
conn.commit()
conn.close()
print(f"🔒 Locked: {notif['id'][:8]}")
def cmd_notif_unlock(args) -> None:
"""Unlock notification after processing completes."""
conn = get_db()
notif = conn.execute(
"SELECT * FROM notifications WHERE id LIKE ?",
(f"{args.id}%",)
).fetchone()
if not notif:
print(f"Error: Notification not found: {args.id}", file=sys.stderr)
sys.exit(1)
conn.execute("UPDATE notifications SET processing = 0 WHERE id = ?", (notif["id"],))
conn.commit()
conn.close()
print(f"🔓 Unlocked: {notif['id'][:8]}")
def cmd_notif_tag(args) -> None:
"""Add a tag to a notification."""
conn = get_db()
_migrate_filter_columns(conn)
notif = conn.execute(
"SELECT * FROM notifications WHERE id LIKE ?",
(f"{args.id}%",)
).fetchone()
if not notif:
print(f"Error: Notification not found: {args.id}", file=sys.stderr)
sys.exit(1)
existing_tags = set(filter(None, (notif["tags"] or "").split(",")))
existing_tags.add(args.tag)
new_tags = ",".join(sorted(existing_tags))
conn.execute("UPDATE notifications SET tags = ? WHERE id = ?", (new_tags, notif["id"]))
conn.commit()
conn.close()
print(f"✓ Tagged {notif['id'][:8]} with '{args.tag}' → [{new_tags}]")
def cmd_notif_untag(args) -> None:
"""Remove a tag from a notification."""
conn = get_db()
_migrate_filter_columns(conn)
notif = conn.execute(
"SELECT * FROM notifications WHERE id LIKE ?",
(f"{args.id}%",)
).fetchone()
if not notif:
print(f"Error: Notification not found: {args.id}", file=sys.stderr)
sys.exit(1)
existing_tags = set(filter(None, (notif["tags"] or "").split(",")))
existing_tags.discard(args.tag)
new_tags = ",".join(sorted(existing_tags))
conn.execute("UPDATE notifications SET tags = ? WHERE id = ?", (new_tags, notif["id"]))
conn.commit()
conn.close()
print(f"✓ Untagged {notif['id'][:8]} '{args.tag}' → [{new_tags}]")
def cmd_notif_check_prior(args) -> None:
"""Check if any notification with the same URL has been triaged before.
Returns exit code 0 if prior engagement found, 1 if not.
Useful for heartbeat triage to avoid duplicate triage comments.
"""
conn = get_db()
_migrate_filter_columns(conn)
url = args.url
# Find any notification with this URL that has been acted on, triaged, or has the 'triaged' tag
prior = conn.execute("""
SELECT id, title, acted_at, action_taken, tags, seen_at
FROM notifications
WHERE url = ? AND (
acted_at IS NOT NULL
OR tags LIKE '%triaged%'
)
ORDER BY seen_at DESC
LIMIT 5
""", (url,)).fetchall()
conn.close()
if prior:
print(f"⚠ PRIOR ENGAGEMENT FOUND for {url}")
for p in prior:
tags = f" [tags: {p['tags']}]" if p["tags"] else ""
action = f" → {p['action_taken']}" if p["action_taken"] else ""
print(f" [{p['id'][:8]}] seen {p['seen_at'][:10]}{action}{tags}")
print(f"\n→ This is a FOLLOW-UP. Do NOT re-triage. Read thread and respond contextually.")
sys.exit(0)
else:
print(f"✓ No prior engagement for {url} — first touch, triage OK.")
sys.exit(1)
def cmd_notif_dismiss(args) -> None:
"""Dismiss notification."""
conn = get_db()
@ -963,14 +795,6 @@ def cmd_notif_dismiss(args) -> None:
WHERE id = ?
""", (f"dismissed: {reason}", notif["id"]))
# Auto-log activity
title = notif["title"] or "-"
detail = f"Dismissed: {reason} — {title}"
conn.execute(
"INSERT INTO activity_log (issue_id, action, detail, github_url) VALUES (?, ?, ?, ?)",
(notif["repo_id"], "dismiss", detail, notif["url"])
)
conn.commit()
conn.close()
@ -998,15 +822,13 @@ def cmd_notif_audit(args) -> None:
print(f" - {n['title'][:50]}...")
# Create escalation
url = n["url"] if "url" in n.keys() and n["url"] else "N/A"
conn.execute("""
INSERT INTO escalations (category, title, detail)
VALUES (?, ?, ?)
""", (
"stale-notification",
f"Stale: {n['title'][:100]}",
f"Notification {n['id']} has been pending for >{stale_hours}h.\nURL: {url}"
f"Notification {n['id']} has been pending for >{stale_hours}h.\nURL: {n.get('url', 'N/A')}"
))
conn.commit()
@ -1285,10 +1107,7 @@ def main():
# Notification commands
notif_parser = subparsers.add_parser("notif", help="Notification management")
notif_sub = notif_parser.add_subparsers(
dest="notif_cmd",
help="pull|list|act|dismiss|lock|unlock|tag|untag|check-prior|audit|filter"
)
notif_sub = notif_parser.add_subparsers(dest="notif_cmd")
pull_parser = notif_sub.add_parser("pull", help="Pull notifications from GitHub/Forgejo")
pull_parser.add_argument("--dry-run", action="store_true", help="Show what would be pulled")
@ -1327,23 +1146,6 @@ def main():
dismiss_parser.add_argument("id", help="Notification ID (prefix match)")
dismiss_parser.add_argument("reason", nargs="?", help="Dismissal reason")
lock_parser = notif_sub.add_parser("lock", help="Lock notification for processing")
lock_parser.add_argument("id", help="Notification ID (prefix match)")
unlock_parser = notif_sub.add_parser("unlock", help="Unlock notification after processing")
unlock_parser.add_argument("id", help="Notification ID (prefix match)")
tag_parser = notif_sub.add_parser("tag", help="Add tag to notification")
tag_parser.add_argument("id", help="Notification ID (prefix match)")
tag_parser.add_argument("tag", help="Tag to add (e.g. triaged, hostile, follow-up)")
untag_parser = notif_sub.add_parser("untag", help="Remove tag from notification")
untag_parser.add_argument("id", help="Notification ID (prefix match)")
untag_parser.add_argument("tag", help="Tag to remove")
check_prior_parser = notif_sub.add_parser("check-prior", help="Check for prior engagement on a URL")
check_prior_parser.add_argument("url", help="Issue/PR URL to check")
audit_parser = notif_sub.add_parser("audit", help="Audit stale notifications")
audit_parser.add_argument("--stale-hours", type=int, help="Hours to consider stale (default: 48)")
@ -1415,16 +1217,6 @@ def main():
cmd_notif_act(args)
elif args.notif_cmd == "dismiss":
cmd_notif_dismiss(args)
elif args.notif_cmd == "lock":
cmd_notif_lock(args)
elif args.notif_cmd == "unlock":
cmd_notif_unlock(args)
elif args.notif_cmd == "tag":
cmd_notif_tag(args)
elif args.notif_cmd == "untag":
cmd_notif_untag(args)
elif args.notif_cmd == "check-prior":
cmd_notif_check_prior(args)
elif args.notif_cmd == "audit":
cmd_notif_audit(args)
elif args.notif_cmd == "filter":

View file

@ -271,7 +271,7 @@
const yesterday = new Date(Date.now() - 86400000).toISOString();
const activity = await fetchData('activity_log',
`SELECT COUNT(*) as c FROM activity_log WHERE timestamp > '${yesterday}' AND action != 'dismiss'`);
`SELECT COUNT(*) as c FROM activity_log WHERE timestamp > '${yesterday}'`);
document.getElementById('activity-count').textContent = activity[0]?.c || 0;
const escalations = await fetchData('escalations',
@ -287,61 +287,42 @@
`(last poll ${timeAgo(p.polled_at)} — ${p.new_count} new)`;
}
// Notifications table — show ALL notifications with action taken
// Notifications table — show pending (non-dismissed) by default
const notifications = await fetchData('notifications',
`SELECT n.*, a.detail as action_detail, a.action as action_type
FROM notifications n
LEFT JOIN activity_log a ON a.github_url = n.url
ORDER BY n.seen_at DESC LIMIT 30`);
'SELECT * FROM notifications WHERE dismissed = 0 ORDER BY seen_at DESC LIMIT 30');
if (notifications.length === 0) {
document.getElementById('notifications-table').innerHTML =
'<div class="loading">No notifications yet. ✨</div>';
'<div class="loading">No pending notifications. ✨</div>';
} else {
let html = `<table>
<thead><tr><th>TIME</th><th>ISSUE</th><th>REASON</th><th>TITLE</th><th>ACTION TAKEN</th></tr></thead>
<thead><tr><th>TIME</th><th>ISSUE</th><th>REASON</th><th>TITLE</th><th>FILTER</th></tr></thead>
<tbody>`;
for (const n of notifications) {
const ownTag = n.is_own_pr ? '🔵 ' : '';
const ciTag = n.ci_status ? ` [CI:${n.ci_status}]` : '';
const repo_short = n.repo_id ? n.repo_id.split('/').pop() : '-';
let issueLabel = repo_short;
if (n.url) {
const m = n.url.match(/\/(pull|issues)\/(\d+)/);
if (m) {
const prefix = m[1] === 'pull' ? 'PR' : 'IS';
issueLabel = `${repo_short} ${prefix} #${m[2]}`;
}
}
const issue = n.url ? `<a href="${n.url}" target="_blank">${issueLabel}</a>` : issueLabel;
const issue = n.url ? `<a href="${n.url}" target="_blank">${repo_short}</a>` : repo_short;
const reasonClass = n.reason === 'mention' ? 'accent' :
n.reason === 'author' ? 'success' :
n.reason === 'review_requested' ? 'warning' : '';
const reasonBadge = `<span class="status-badge ${reasonClass ? 'status-' + reasonClass : ''}" style="background:var(--${reasonClass || 'text-dim'}); color:#000; padding:2px 8px; border-radius:4px; font-size:0.8em">${n.reason}</span>`;
// Action taken column
let actionText = '';
if (n.action_detail) {
actionText = n.action_detail;
} else if (n.dismissed && n.filter_reason) {
actionText = `dismissed: ${n.filter_reason.replace(/^KEEP \(rule\): /, '')}`;
} else if (!n.dismissed) {
actionText = '<span style="color:var(--warning)">pending</span>';
}
html += `<tr style="${n.dismissed ? 'opacity:0.7' : ''}">
const filterShort = (n.filter_reason || '').replace(/^KEEP: /, '');
html += `<tr>
<td class="timestamp">${timeAgo(n.seen_at)}</td>
<td>${issue}</td>
<td>${reasonBadge}</td>
<td>${ownTag}${n.title || '-'}${ciTag}</td>
<td style="font-size:0.85em">${actionText}</td>
<td style="color:var(--text-dim);font-size:0.85em">${filterShort}</td>
</tr>`;
}
html += '</tbody></table>';
document.getElementById('notifications-table').innerHTML = html;
}
// Activity table — filter out dismissals, show only real actions
// Activity table
const activityLog = await fetchData('activity_log',
"SELECT * FROM activity_log WHERE action != 'dismiss' ORDER BY timestamp DESC LIMIT 10");
'SELECT * FROM activity_log ORDER BY timestamp DESC LIMIT 10');
if (activityLog.length === 0) {
document.getElementById('activity-table').innerHTML =

View file

@ -1,75 +0,0 @@
# Agent Behavioral Guidelines
Rules for AI agents using Argus for GitHub/Forgejo community management.
These guidelines govern **how** agents interact with humans on GitHub — not the mechanics of Argus itself (see [SKILL.md](SKILL.md) for that).
## Core Principle
**You represent a project.** Every comment you post reflects on the maintainers. Be helpful, be accurate, be kind — or be silent.
## Engagement Rules
### Always Respond To
- Direct mentions (`@your-username`)
- Review requests on PRs you're assigned to
- Questions on issues you previously commented on
### Never Respond To
- **Hostile or abusive comments.** Dismiss the notification, alert your operator. Do not engage. Do not defend yourself. Silence starves trolls.
- Conversations where you have nothing substantive to add
- Heated arguments between humans — stay out
### Be Careful With
- Closing issues — always explain why, link duplicates
- Marking things as "won't fix" — defer to human maintainers
- Making promises about timelines or features
- Speculation about root causes without evidence
## Tone & Style
- **Be concise.** Developers read hundreds of notifications. Respect their time.
- **Be specific.** "This might be a problem" is useless. "Line 42 in `rpc.py` catches `Exception` instead of `ConnectionError`" is useful.
- **Be humble.** Say "I think" not "This is." You're an AI — you can be wrong.
- **No filler.** Skip "Great question!", "Thanks for reporting!", "I'd be happy to help!" — just help.
- **Use code.** Show diffs, snippets, reproduction steps. Words explain; code proves.
## Triage Quality
When assessing issues:
1. **Read the full thread** before commenting — someone may have already answered
2. **Reproduce if possible** — or explain why you can't
3. **Classify honestly** — don't mark "NEEDS-INFO" to avoid doing work
4. **Link related issues** — duplicates, upstream bugs, related PRs
5. **Propose concrete fixes** — file paths, function names, approach. Vague assessments waste everyone's time.
## Escalation
When in doubt, escalate to your operator rather than posting something wrong or inappropriate. A missed comment is recoverable. A bad comment lives forever.
### Escalate When
- You're unsure if a response is appropriate
- The issue involves security, legal, or sensitive topics
- A human is clearly frustrated and needs a human response
- You'd need to make a judgment call about project direction
## PR Reviews
- **Review the diff, not just the description**
- **Test mentally** — trace the code path, check edge cases
- **Check scope** — flag if changes exceed what the PR claims
- **Be constructive** — suggest alternatives, don't just criticize
- **Approve only when confident** — "LGTM" from an AI that didn't actually verify is worse than no review
## Learning From Mistakes
When you make a mistake in a community interaction:
1. Document it (postmortem, memory, or learning log)
2. Update these guidelines if a new rule is needed
3. Alert your operator
4. If the mistake was public: correct it publicly, briefly, without drama
---
*These guidelines are living. Update them as you learn.*

View file

@ -40,8 +40,6 @@ CREATE TABLE IF NOT EXISTS notifications (
filter_reason TEXT,
is_own_pr INTEGER DEFAULT 0,
ci_status TEXT,
processing INTEGER DEFAULT 0,
tags TEXT DEFAULT '',
FOREIGN KEY (issue_id) REFERENCES issues(id),
FOREIGN KEY (repo_id) REFERENCES repos(id)
);

View file

@ -1,106 +0,0 @@
import sqlite3
import tempfile
import unittest
from importlib.machinery import SourceFileLoader
from pathlib import Path
from types import SimpleNamespace
ARGUS_PATH = Path(__file__).resolve().parents[1] / "bin" / "argus"
argus = SourceFileLoader("argus_cli", str(ARGUS_PATH)).load_module()
def init_db(db_path: str):
conn = sqlite3.connect(db_path)
conn.execute(
"""
CREATE TABLE notifications (
id TEXT PRIMARY KEY,
title TEXT NOT NULL,
url TEXT,
dismissed INTEGER DEFAULT 0,
acted_at TEXT,
seen_at TEXT
)
"""
)
conn.execute(
"""
CREATE TABLE escalations (
id INTEGER PRIMARY KEY AUTOINCREMENT,
category TEXT,
title TEXT,
detail TEXT
)
"""
)
conn.commit()
conn.close()
class TestNotifAudit(unittest.TestCase):
def test_audit_handles_sqlite_row_without_get(self):
with tempfile.TemporaryDirectory() as td:
db_path = str(Path(td) / "argus.db")
init_db(db_path)
conn = sqlite3.connect(db_path)
conn.execute(
"""
INSERT INTO notifications (id, title, url, dismissed, acted_at, seen_at)
VALUES (?, ?, ?, 0, NULL, '2000-01-01T00:00:00')
""",
("n1", "old stale notification", None),
)
conn.commit()
conn.close()
def fake_get_db():
c = sqlite3.connect(db_path)
c.row_factory = sqlite3.Row
return c
argus.get_db = fake_get_db
argus.cmd_notif_audit(SimpleNamespace(stale_hours=4))
conn = sqlite3.connect(db_path)
row = conn.execute("SELECT category, detail FROM escalations").fetchone()
conn.close()
self.assertIsNotNone(row)
self.assertEqual(row[0], "stale-notification")
self.assertIn("Notification n1 has been pending", row[1])
def test_audit_no_stale_notifications_creates_no_escalation(self):
with tempfile.TemporaryDirectory() as td:
db_path = str(Path(td) / "argus.db")
init_db(db_path)
conn = sqlite3.connect(db_path)
conn.execute(
"""
INSERT INTO notifications (id, title, url, dismissed, acted_at, seen_at)
VALUES (?, ?, ?, 0, NULL, '2999-01-01T00:00:00')
""",
("n2", "fresh notification", "https://example.com"),
)
conn.commit()
conn.close()
def fake_get_db():
c = sqlite3.connect(db_path)
c.row_factory = sqlite3.Row
return c
argus.get_db = fake_get_db
argus.cmd_notif_audit(SimpleNamespace(stale_hours=4))
conn = sqlite3.connect(db_path)
count = conn.execute("SELECT COUNT(*) FROM escalations").fetchone()[0]
conn.close()
self.assertEqual(count, 0)
if __name__ == "__main__":
unittest.main()