Compare commits
2 Commits
d2372a98fa
...
4eecfc4aff
| Author | SHA1 | Date | |
|---|---|---|---|
| 4eecfc4aff | |||
| ce5b31485c |
@@ -162,34 +162,6 @@ def get_next_scheduled_event(now: datetime | None = None) -> dict:
|
||||
return {"at": next_dt, "type": next_type}
|
||||
|
||||
|
||||
def start_dashboard() -> None:
|
||||
"""Start the minimal dashboard in a background thread."""
|
||||
enabled = os.getenv("DASHBOARD_ENABLED", "1").strip().lower() not in {
|
||||
"0", "false", "no"}
|
||||
if not enabled:
|
||||
return
|
||||
|
||||
host = os.getenv("DASHBOARD_HOST", "0.0.0.0")
|
||||
port = int(os.getenv("DASHBOARD_PORT", "8080"))
|
||||
|
||||
def _run() -> None:
|
||||
try:
|
||||
from dashboard import create_app
|
||||
|
||||
app = create_app(
|
||||
get_state=get_state_snapshot,
|
||||
get_next_event=lambda: get_next_scheduled_event(),
|
||||
)
|
||||
app.config["TEMPLATES_PATH"] = os.getenv(
|
||||
"TEMPLATES_PATH", "templates.json")
|
||||
app.run(host=host, port=port, debug=False, use_reloader=False)
|
||||
except Exception as e:
|
||||
logging.error(f"Dashboard failed to start: {e}")
|
||||
|
||||
thread = threading.Thread(target=_run, name="dashboard", daemon=True)
|
||||
thread.start()
|
||||
|
||||
|
||||
def load_timezones() -> list[dict]:
|
||||
"""Load timezones from csv file."""
|
||||
# Read the CSV file and return a list of timezones
|
||||
@@ -302,11 +274,11 @@ def should_delete_message(
|
||||
message: dict,
|
||||
webhook_id: str,
|
||||
author_id: str,
|
||||
five_minutes_ago_timestamp: int,
|
||||
cutoff: int,
|
||||
) -> bool:
|
||||
message_timestamp = int(parse_message_timestamp(message).timestamp())
|
||||
return (
|
||||
message_timestamp <= five_minutes_ago_timestamp
|
||||
message_timestamp <= cutoff
|
||||
and message.get("webhook_id") == webhook_id
|
||||
and message.get("author", {}).get("id") == author_id
|
||||
)
|
||||
@@ -744,16 +716,14 @@ def main() -> None:
|
||||
"""
|
||||
Main function to run the scheduler.
|
||||
"""
|
||||
# start_dashboard()
|
||||
|
||||
# Schedule notifications
|
||||
schedule.every().hour.at(":15").do(send_notification, "reminder")
|
||||
schedule.every().hour.at(":20").do(send_notification, "420")
|
||||
# schedule.every().hour.at(":45").do(send_notification, "reminder_halftime")
|
||||
# schedule.every().hour.at(":50").do(send_notification, "halftime")
|
||||
|
||||
# Schedule deletion of old messages every 3 minutes
|
||||
schedule.every(3).minutes.do(delete_old_messages, 6)
|
||||
# Schedule deletion of old messages every 5 minutes
|
||||
schedule.every(5).minutes.do(delete_old_messages, 6)
|
||||
|
||||
logging.info("Scheduler started.")
|
||||
|
||||
@@ -761,7 +731,7 @@ def main() -> None:
|
||||
# send_notification("420")
|
||||
|
||||
# delete old messages on startup to clean up any previous notifications
|
||||
delete_old_messages(60)
|
||||
# delete_old_messages(6)
|
||||
|
||||
try:
|
||||
while True:
|
||||
|
||||
@@ -1,65 +0,0 @@
|
||||
from datetime import datetime
|
||||
|
||||
from dashboard import create_app
|
||||
|
||||
|
||||
def test_dashboard_renders_locations():
|
||||
state = {
|
||||
"running": True,
|
||||
"started_at": datetime(2025, 1, 1, 0, 0, 0),
|
||||
"last_type": "420",
|
||||
"last_attempt_at": datetime(2025, 1, 1, 0, 1, 2),
|
||||
"last_success_at": datetime(2025, 1, 1, 0, 1, 3),
|
||||
"last_status_code": 204,
|
||||
"last_error": None,
|
||||
"last_locations": ["Nowhere", "Somewhere"],
|
||||
}
|
||||
|
||||
app = create_app(
|
||||
get_state=lambda: state,
|
||||
get_next_event=lambda: {"type": "reminder", "at": datetime(2025, 1, 1, 0, 15, 0)},
|
||||
)
|
||||
|
||||
client = app.test_client()
|
||||
resp = client.get("/")
|
||||
assert resp.status_code == 200
|
||||
body = resp.data.decode("utf-8")
|
||||
assert "thc-webhook" in body
|
||||
assert "Locations" in body
|
||||
assert "<li>Nowhere</li>" in body
|
||||
assert "<li>Somewhere</li>" in body
|
||||
|
||||
|
||||
def test_admin_roundtrip(tmp_path, monkeypatch):
|
||||
templates_path = tmp_path / "templates.json"
|
||||
monkeypatch.setenv("TEMPLATES_PATH", str(templates_path))
|
||||
|
||||
app = create_app(
|
||||
get_state=lambda: {},
|
||||
get_next_event=lambda: {"type": "reminder", "at": datetime(2025, 1, 1, 0, 15, 0)},
|
||||
)
|
||||
app.config["TEMPLATES_PATH"] = str(templates_path)
|
||||
client = app.test_client()
|
||||
|
||||
# GET admin should render
|
||||
resp = client.get("/admin")
|
||||
assert resp.status_code == 200
|
||||
|
||||
# POST should save
|
||||
form = {
|
||||
"reminder__text": "R",
|
||||
"reminder__color": "#e67e22",
|
||||
"reminder__image_url": "",
|
||||
"420__text": "B",
|
||||
"420__color": "0x2ecc71",
|
||||
"420__image_url": "http://example.com/img.png",
|
||||
"reminder_halftime__text": "H",
|
||||
"reminder_halftime__color": "15158306",
|
||||
"reminder_halftime__image_url": "",
|
||||
"halftime__text": "HT",
|
||||
"halftime__color": "3066993",
|
||||
"halftime__image_url": "",
|
||||
}
|
||||
resp = client.post("/admin", data=form)
|
||||
assert resp.status_code == 200
|
||||
assert templates_path.exists()
|
||||
Reference in New Issue
Block a user