mirror of
https://github.com/alexgetmancom/miband-bot.git
synced 2026-09-08 11:34:07 +03:00
rewrite bot in TypeScript and add locale switching
This commit is contained in:
Alex Getman
@@ -0,0 +1,149 @@
|
||||
import { describe, expect, test } from "bun:test";
|
||||
import { mkdtempSync, rmSync, writeFileSync } from "node:fs";
|
||||
import { join } from "node:path";
|
||||
import { createBot } from "../src/bot/app.js";
|
||||
import type { Locale } from "../src/bot/i18n.js";
|
||||
import { type AppConfig, loadConfig } from "../src/config.js";
|
||||
import { initHealthDb, initStateDb, setUserLocale, withHealthDb } from "../src/storage/health.js";
|
||||
|
||||
type ApiCall = { method: string; payload: Record<string, unknown> };
|
||||
type KeyboardButton = { text: string; callback_data?: string };
|
||||
|
||||
function callbackUpdate(data: string): Parameters<ReturnType<typeof createBot>["handleUpdate"]>[0] {
|
||||
return {
|
||||
update_id: Date.now(),
|
||||
callback_query: {
|
||||
id: `query-${Date.now()}`,
|
||||
data,
|
||||
from: { id: 42, is_bot: false, first_name: "Test" },
|
||||
chat_instance: "test",
|
||||
message: {
|
||||
message_id: 99,
|
||||
date: Math.floor(Date.now() / 1000),
|
||||
chat: { id: 42, type: "private", first_name: "Test" },
|
||||
text: "menu",
|
||||
},
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
function configFor(dir: string): AppConfig {
|
||||
return loadConfig({
|
||||
TELEGRAM_BOT_TOKEN: "123:abc",
|
||||
TELEGRAM_ALLOWED_USER_IDS: "42,43",
|
||||
DATA_DIR: dir,
|
||||
});
|
||||
}
|
||||
|
||||
async function runCallback(
|
||||
data: string,
|
||||
language: Locale = "en",
|
||||
seed?: (config: AppConfig) => void,
|
||||
): Promise<ApiCall[]> {
|
||||
const dir = mkdtempSync(join(process.env.TMPDIR ?? "/tmp", "miband-ui-"));
|
||||
try {
|
||||
const config = configFor(dir);
|
||||
writeFileSync(join(dir, "token_42.json"), "{}");
|
||||
initStateDb(config.botStateDbPath);
|
||||
setUserLocale(config, 42, language);
|
||||
initHealthDb(join(dir, "miband_42.db"));
|
||||
initHealthDb(join(dir, "miband_43.db"));
|
||||
seed?.(config);
|
||||
const calls: ApiCall[] = [];
|
||||
const bot = createBot(config);
|
||||
bot.botInfo = { id: 999, is_bot: true, first_name: "Test", username: "test_bot" } as never;
|
||||
bot.api.config.use(async (_previous, method, payload) => {
|
||||
calls.push({ method: String(method), payload: payload as Record<string, unknown> });
|
||||
return { ok: true, result: true } as never;
|
||||
});
|
||||
await bot.handleUpdate(callbackUpdate(data));
|
||||
return calls;
|
||||
} finally {
|
||||
rmSync(dir, { recursive: true, force: true });
|
||||
}
|
||||
}
|
||||
|
||||
function editCall(calls: ApiCall[]): ApiCall {
|
||||
const call = calls.find((item) => item.method === "editMessageText");
|
||||
if (!call) throw new Error("editMessageText call was not made");
|
||||
return call;
|
||||
}
|
||||
|
||||
function rows(call: ApiCall): KeyboardButton[][] {
|
||||
const markup = call.payload.reply_markup as { inline_keyboard: KeyboardButton[][] };
|
||||
return markup.inline_keyboard;
|
||||
}
|
||||
|
||||
function button(rowsToRead: KeyboardButton[][], text: string): KeyboardButton {
|
||||
const found = rowsToRead.flat().find((item) => item.text === text);
|
||||
if (!found) throw new Error(`Button not found: ${text}`);
|
||||
return found;
|
||||
}
|
||||
|
||||
describe("bot UI callback parity", () => {
|
||||
test("opens versus from its legacy menu callback", async () => {
|
||||
const call = editCall(await runCallback("menu:versus"));
|
||||
expect(call.payload.text).toBe("📊 <b>Compare activity:</b>");
|
||||
expect(button(rows(call), "Today").callback_data).toBe("versus:1");
|
||||
expect(button(rows(call), "📊 Weekly").callback_data).toBe("versus:7");
|
||||
});
|
||||
|
||||
test("preserves the family return destination", async () => {
|
||||
const weekly = editCall(await runCallback("menu:family:weekly"));
|
||||
expect(button(rows(weekly), "⬅️ Back").callback_data).toBe("menu:weekly_back");
|
||||
|
||||
const trends = editCall(await runCallback("menu:family:trends"));
|
||||
expect(button(rows(trends), "⬅️ Back").callback_data).toBe("menu:trends");
|
||||
|
||||
const weeklyBack = editCall(await runCallback("menu:weekly_back"));
|
||||
expect(String(weeklyBack.payload.text)).toContain("📊 Weekly summary:");
|
||||
expect(button(rows(weeklyBack), "👪 Family").callback_data).toBe("menu:family:weekly");
|
||||
});
|
||||
|
||||
test("marks the selected trends period", async () => {
|
||||
const call = editCall(await runCallback("period:30d"));
|
||||
expect(button(rows(call), "· 30 days ·").callback_data).toBe("period:30d");
|
||||
expect(button(rows(call), "All time").callback_data).toBe("period:all");
|
||||
});
|
||||
|
||||
test("uses English as the default main keyboard", async () => {
|
||||
const call = editCall(await runCallback("menu:main"));
|
||||
expect(button(rows(call), "😴 Sleep").callback_data).toBe("menu:sleep");
|
||||
expect(button(rows(call), "📊 Weekly").callback_data).toBe("menu:trends");
|
||||
expect(button(rows(call), "⚙️ Settings").callback_data).toBe("menu:more");
|
||||
});
|
||||
|
||||
test("offers Russian and Spanish in the language settings", async () => {
|
||||
const languageMenu = editCall(await runCallback("menu:language"));
|
||||
expect(button(rows(languageMenu), "Русский").callback_data).toBe("locale:ru");
|
||||
expect(button(rows(languageMenu), "Español").callback_data).toBe("locale:es");
|
||||
|
||||
const spanish = editCall(await runCallback("locale:es"));
|
||||
expect(button(rows(spanish), "· Español ·").callback_data).toBe("locale:es");
|
||||
const spanishMain = editCall(await runCallback("menu:main", "es"));
|
||||
expect(button(rows(spanishMain), "😴 Sueño").callback_data).toBe("menu:sleep");
|
||||
});
|
||||
|
||||
test("renders Spanish trend labels", async () => {
|
||||
const call = editCall(await runCallback("menu:trends", "es"));
|
||||
expect(String(call.payload.text)).toContain("📊 Tendencias");
|
||||
expect(button(rows(call), "30 días").callback_data).toBe("period:30d");
|
||||
expect(button(rows(call), "👪 Familia").callback_data).toBe("menu:family:trends");
|
||||
});
|
||||
|
||||
test("does not append /100 to an unavailable sleep score", async () => {
|
||||
const call = editCall(
|
||||
await runCallback("menu:sleep", "ru", (config) => {
|
||||
withHealthDb(config, 42, (database) => {
|
||||
database
|
||||
.prepare(
|
||||
"INSERT INTO sleep_daily (date,light_sleep_min,deep_sleep_min,start_time,end_time,total_duration_min,sleep_score) VALUES (?,?,?,?,?,?,?)",
|
||||
)
|
||||
.run("2026-08-12", 120, 120, 1_755_000_000, 1_755_025_200, 240, 0);
|
||||
});
|
||||
}),
|
||||
);
|
||||
expect(String(call.payload.text)).toContain("Качество <b>н/д</b>");
|
||||
expect(String(call.payload.text)).not.toContain("Качество <b>н/д / 100</b>");
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,34 @@
|
||||
import { describe, expect, test } from "bun:test";
|
||||
import { ConfigurationError, loadConfig } from "../src/config.js";
|
||||
|
||||
describe("loadConfig", () => {
|
||||
test("keeps http-only mode usable without Telegram credentials", () => {
|
||||
const config = loadConfig({ BOT_MODE: "http-only", DATABASE_URL: ":memory:", QUERY_DURATION: "2" });
|
||||
expect(config.BOT_MODE).toBe("http-only");
|
||||
expect(config.QUERY_DURATION).toBe(2);
|
||||
expect(config.dataDir).toContain("data");
|
||||
});
|
||||
|
||||
test("parses the personal allowlist", () => {
|
||||
const config = loadConfig({ TELEGRAM_BOT_TOKEN: "123:abc", TELEGRAM_ALLOWED_USER_IDS: "42, 7" });
|
||||
expect(config.allowedUserIds).toEqual([42, 7]);
|
||||
});
|
||||
|
||||
test("rejects malformed IDs and incomplete webhook settings", () => {
|
||||
expect(() => loadConfig({ TELEGRAM_BOT_TOKEN: "123:abc", TELEGRAM_ALLOWED_USER_IDS: "42,nope" })).toThrow(
|
||||
ConfigurationError,
|
||||
);
|
||||
expect(() =>
|
||||
loadConfig({ TELEGRAM_BOT_TOKEN: "123:abc", BOT_MODE: "webhook", TELEGRAM_ALLOWED_USER_IDS: "42" }),
|
||||
).toThrow(ConfigurationError);
|
||||
});
|
||||
|
||||
test("treats empty strings as unset", () => {
|
||||
expect(loadConfig({ BOT_MODE: "http-only", TELEGRAM_BOT_TOKEN: "" }).TELEGRAM_BOT_TOKEN).toBeUndefined();
|
||||
});
|
||||
|
||||
test("parses textual booleans instead of coercing false to true", () => {
|
||||
const config = loadConfig({ BOT_MODE: "http-only", ENABLE_FDS_SLEEP_DETAILS: "false" });
|
||||
expect(config.ENABLE_FDS_SLEEP_DETAILS).toBe(false);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,8 @@
|
||||
import { describe, expect, test } from "bun:test";
|
||||
import { parseAllDaySleepBytes } from "../src/xiaomi/fds.js";
|
||||
|
||||
describe("FDS parser", () => {
|
||||
test("rejects truncated sleep payloads", () => {
|
||||
expect(parseAllDaySleepBytes(new Uint8Array([0, 0, 0, 0, 0, 1, 0, 0, 0]))).toBeNull();
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,10 @@
|
||||
import { describe, expect, test } from "bun:test";
|
||||
import { esc, minutes, stepBar } from "../src/bot/formatting.js";
|
||||
|
||||
describe("bot formatting", () => {
|
||||
test("escapes Telegram HTML and formats health values", () => {
|
||||
expect(esc('<secret> & "value"')).toBe("<secret> & "value"");
|
||||
expect(minutes(396)).toBe("6 h 36 min");
|
||||
expect(stepBar(5000)).toContain("50%");
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,33 @@
|
||||
import { describe, expect, test } from "bun:test";
|
||||
import { mkdtempSync, rmSync } from "node:fs";
|
||||
import { join } from "node:path";
|
||||
import { initHealthDb, zipExport } from "../src/storage/health.js";
|
||||
import { openDatabase } from "../src/storage/kv.js";
|
||||
|
||||
function tempDir(): string {
|
||||
return mkdtempSync(join(process.env.TMPDIR ?? "/tmp", "miband-ts-"));
|
||||
}
|
||||
|
||||
describe("health storage", () => {
|
||||
test("creates the health schema and exports CSV files as ZIP", async () => {
|
||||
const dir = tempDir();
|
||||
const dbPath = join(dir, "miband_42.db");
|
||||
initHealthDb(dbPath);
|
||||
const database = openDatabase(dbPath);
|
||||
database.sqlite.query("INSERT INTO steps_daily (date,total_steps) VALUES (?,?)").run("2026-08-11", 1234);
|
||||
database.close();
|
||||
const config = {
|
||||
dataDir: dir,
|
||||
dbPath,
|
||||
statusPath: join(dir, "status.json"),
|
||||
botStateDbPath: join(dir, "state.db"),
|
||||
allowedUserIds: [42],
|
||||
BOT_MODE: "http-only",
|
||||
QUERY_DURATION: 2,
|
||||
} as never;
|
||||
const archive = await zipExport(config, 42);
|
||||
expect(archive).not.toBeNull();
|
||||
expect(new Uint8Array(archive ?? []).length).toBeGreaterThan(0);
|
||||
rmSync(dir, { recursive: true, force: true });
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,14 @@
|
||||
import { describe, expect, test } from "bun:test";
|
||||
import { mkdtempSync, rmSync } from "node:fs";
|
||||
import { join } from "node:path";
|
||||
import { withExclusiveFileLock } from "../src/storage/secure-files.js";
|
||||
|
||||
describe("file locks", () => {
|
||||
test("releases the lock directory after the action", async () => {
|
||||
const directory = mkdtempSync(join(process.env.TMPDIR ?? "/tmp", "miband-lock-"));
|
||||
const path = join(directory, "sync.lock");
|
||||
await withExclusiveFileLock(path, async () => "first");
|
||||
await expect(withExclusiveFileLock(path, async () => "second")).resolves.toBe("second");
|
||||
rmSync(directory, { recursive: true, force: true });
|
||||
});
|
||||
});
|
||||
@@ -1,286 +0,0 @@
|
||||
# SPDX-License-Identifier: GPL-3.0-or-later
|
||||
# Copyright (C) 2026 Alexey
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import os
|
||||
import time
|
||||
from pathlib import Path
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
import pytest
|
||||
|
||||
import miband_tracker.bot.app as bot_app
|
||||
from miband_tracker import storage
|
||||
from miband_tracker.config import Settings
|
||||
from miband_tracker.sync import SyncResult
|
||||
|
||||
|
||||
class FakeUser:
|
||||
id = 123
|
||||
|
||||
|
||||
class FakeUpdate:
|
||||
effective_user = FakeUser()
|
||||
message = object()
|
||||
callback_query = None
|
||||
|
||||
|
||||
def _settings(tmp_path: Path) -> Settings:
|
||||
return Settings(
|
||||
data_dir=tmp_path,
|
||||
db_path=tmp_path / "miband.db",
|
||||
status_path=tmp_path / "status.json",
|
||||
bot_state_db_path=tmp_path / "fitness_bot_state.db",
|
||||
telegram_bot_token="token",
|
||||
telegram_allowed_user_id=123,
|
||||
sync_interval=900,
|
||||
query_duration=2,
|
||||
enable_fds_sleep_details=True,
|
||||
)
|
||||
|
||||
|
||||
def test_service_menu_does_not_expose_invites_or_second_user(
|
||||
tmp_path: Path,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
settings = _settings(tmp_path)
|
||||
|
||||
monkeypatch.setattr(bot_app, "SETTINGS", settings)
|
||||
monkeypatch.setattr(bot_app, "ALLOWED_USER_ID", 123)
|
||||
monkeypatch.setattr(bot_app, "DB_PATH", str(settings.canonical_user_db_path()))
|
||||
|
||||
keyboard = bot_app.more_keyboard()
|
||||
labels = [
|
||||
button.text
|
||||
for row in keyboard.inline_keyboard
|
||||
for button in row
|
||||
]
|
||||
rendered = "\n".join([*labels, bot_app.more_text()])
|
||||
|
||||
assert "Принять приглашения" not in rendered
|
||||
assert "Приглашения" not in rendered
|
||||
assert "втор" not in rendered.lower()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_cmd_start_without_token_shows_onboarding(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
update = FakeUpdate()
|
||||
context = object()
|
||||
show_onboarding = AsyncMock()
|
||||
show_main_menu = AsyncMock()
|
||||
|
||||
monkeypatch.setattr(bot_app, "is_allowed", lambda _: True)
|
||||
monkeypatch.setattr(bot_app, "has_xiaomi_token", lambda: False)
|
||||
monkeypatch.setattr(bot_app, "safe_delete", AsyncMock())
|
||||
monkeypatch.setattr(bot_app, "show_onboarding", show_onboarding)
|
||||
monkeypatch.setattr(bot_app, "show_main_menu", show_main_menu)
|
||||
|
||||
await bot_app.cmd_start(update, context)
|
||||
|
||||
show_onboarding.assert_awaited_once_with(update, context, force_new=True)
|
||||
show_main_menu.assert_not_awaited()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_start_xiaomi_login_saves_token_syncs_and_opens_menu(
|
||||
tmp_path: Path,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
update = FakeUpdate()
|
||||
context = object()
|
||||
token_path = tmp_path / "token_123.json"
|
||||
|
||||
class FakeToken:
|
||||
def model_dump(self):
|
||||
return {
|
||||
"user_id": "456",
|
||||
"c_user_id": "",
|
||||
"service_token": "service",
|
||||
"ssecurity": "sec",
|
||||
"pass_token": "pass",
|
||||
"device_id": "device",
|
||||
}
|
||||
|
||||
class FakeAuth:
|
||||
async def login_qr(self, *, qr_callback, max_wait):
|
||||
await qr_callback("https://example.test/qr.png", "https://example.test/login")
|
||||
return FakeToken()
|
||||
|
||||
async def close(self):
|
||||
return None
|
||||
|
||||
settings = _settings(tmp_path)
|
||||
run_sync = AsyncMock(return_value=SyncResult(True, user_id=123))
|
||||
show_main_menu = AsyncMock()
|
||||
|
||||
monkeypatch.setattr(bot_app, "SETTINGS", settings)
|
||||
monkeypatch.setattr(bot_app, "ALLOWED_USER_ID", 123)
|
||||
monkeypatch.setattr(bot_app, "AUTH_LOCK", asyncio.Lock())
|
||||
monkeypatch.setattr(bot_app, "SYNC_LOCK", asyncio.Lock())
|
||||
monkeypatch.setattr(bot_app, "XiaomiAuth", FakeAuth)
|
||||
monkeypatch.setattr(bot_app, "update_menu", AsyncMock())
|
||||
monkeypatch.setattr(bot_app, "run_sync", run_sync)
|
||||
monkeypatch.setattr(bot_app, "show_main_menu", show_main_menu)
|
||||
|
||||
await bot_app.start_xiaomi_login(update, context)
|
||||
|
||||
data = json.loads(token_path.read_text(encoding="utf-8"))
|
||||
assert data["user_id"] == "456"
|
||||
assert data["service_token"] == "service"
|
||||
run_sync.assert_awaited_once()
|
||||
call_args = run_sync.await_args[0]
|
||||
assert call_args[0] == 123
|
||||
assert call_args[1].query_duration == 30
|
||||
show_main_menu.assert_awaited_once_with(update, context)
|
||||
|
||||
|
||||
def test_main_menu_renders_extended_metrics(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
settings = _settings(tmp_path)
|
||||
db_path = settings.canonical_user_db_path()
|
||||
storage.init_health_db(db_path)
|
||||
with storage.sqlite_conn(db_path, row_factory=False) as conn:
|
||||
conn.execute(
|
||||
"INSERT INTO steps_daily (date, total_steps, calories, distance_m, last_sync) VALUES (?, ?, ?, ?, ?)",
|
||||
("2026-05-26", 451, 22.0, 296.0, 1),
|
||||
)
|
||||
conn.execute("INSERT INTO heart_rate (timestamp, value) VALUES (?, ?)", (1779768600, 73))
|
||||
conn.execute("INSERT INTO blood_oxygen (timestamp, spo2, type) VALUES (?, ?, ?)", (1779768900, 99.0, "latest"))
|
||||
conn.execute("INSERT INTO stress (timestamp, value) VALUES (?, ?)", (1779757800, 29))
|
||||
conn.execute(
|
||||
"""
|
||||
INSERT INTO calories_daily
|
||||
(date, total_cal, active_cal, valid_stand_hours, intensity_minutes, last_sync)
|
||||
VALUES (?, ?, ?, ?, ?, ?)
|
||||
""",
|
||||
("2026-05-26", 55.0, None, 1, 6, 1),
|
||||
)
|
||||
conn.commit()
|
||||
|
||||
monkeypatch.setattr(bot_app, "SETTINGS", settings)
|
||||
monkeypatch.setattr(bot_app, "ALLOWED_USER_ID", 123)
|
||||
monkeypatch.setattr(bot_app, "DB_PATH", str(db_path))
|
||||
|
||||
rendered = bot_app.main_menu_text()
|
||||
|
||||
assert "451" in rendered
|
||||
assert "99%" in rendered
|
||||
assert "🧘" in rendered
|
||||
assert "22</b> ккал" in rendered
|
||||
|
||||
|
||||
def test_workouts_text_renders_recent_workout(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
settings = _settings(tmp_path)
|
||||
db_path = settings.canonical_user_db_path()
|
||||
storage.init_health_db(db_path)
|
||||
with storage.sqlite_conn(db_path, row_factory=False) as conn:
|
||||
conn.execute(
|
||||
"""
|
||||
INSERT INTO workouts
|
||||
(workout_id, sport_type, start_time, end_time, duration_sec,
|
||||
calories, avg_hr, max_hr, min_hr, watermark, raw_json)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
""",
|
||||
("w1", "free_training", 1779752903, 1779753050, 142, 7.0, 95, 152, 80, 1, "{}"),
|
||||
)
|
||||
conn.commit()
|
||||
|
||||
monkeypatch.setattr(bot_app, "SETTINGS", settings)
|
||||
monkeypatch.setattr(bot_app, "ALLOWED_USER_ID", 123)
|
||||
monkeypatch.setattr(bot_app, "DB_PATH", str(db_path))
|
||||
|
||||
rendered = bot_app.workouts_text()
|
||||
|
||||
assert "Тренировки" in rendered
|
||||
assert "Свободная" in rendered
|
||||
assert "95 bpm" in rendered
|
||||
|
||||
|
||||
def test_sleep_text_has_no_fds_hint_or_calendar_button(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
settings = _settings(tmp_path)
|
||||
db_path = settings.canonical_user_db_path()
|
||||
storage.init_health_db(db_path)
|
||||
with storage.sqlite_conn(db_path, row_factory=False) as conn:
|
||||
conn.execute(
|
||||
"""
|
||||
INSERT INTO sleep_daily
|
||||
(date, light_sleep_min, deep_sleep_min, start_time, end_time,
|
||||
rem_sleep_min, awake_min, total_duration_min, sleep_score)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
""",
|
||||
("2026-05-26", 137, 123, 1779744540, 1779770460, 136, 0, 396, 64),
|
||||
)
|
||||
conn.commit()
|
||||
|
||||
monkeypatch.setattr(bot_app, "SETTINGS", settings)
|
||||
monkeypatch.setattr(bot_app, "ALLOWED_USER_ID", 123)
|
||||
monkeypatch.setattr(bot_app, "DB_PATH", str(db_path))
|
||||
|
||||
rendered = bot_app.latest_sleep_text()
|
||||
keyboard_labels = [
|
||||
button.text
|
||||
for row in bot_app.back_keyboard().inline_keyboard
|
||||
for button in row
|
||||
]
|
||||
|
||||
assert "Детали подтягиваются" not in rendered
|
||||
assert "FDS" not in rendered
|
||||
assert "Календарь" not in keyboard_labels
|
||||
|
||||
|
||||
def test_trends_screen_labels_and_keyboard(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
settings = _settings(tmp_path)
|
||||
|
||||
monkeypatch.setattr(bot_app, "SETTINGS", settings)
|
||||
monkeypatch.setattr(bot_app, "ALLOWED_USER_ID", 123)
|
||||
monkeypatch.setattr(bot_app, "DB_PATH", str(settings.canonical_user_db_path()))
|
||||
|
||||
rendered = bot_app.period_text(3650)
|
||||
keyboard = bot_app.trends_keyboard(3650)
|
||||
labels = [
|
||||
button.text
|
||||
for row in keyboard.inline_keyboard
|
||||
for button in row
|
||||
]
|
||||
|
||||
assert "Тренды" in rendered
|
||||
assert "Все время" in rendered
|
||||
assert not any("Аналитика" in label for label in labels)
|
||||
assert not any("Детали сна" in label for label in labels)
|
||||
assert any("Все время" in label for label in labels)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_auto_refresh_updates_saved_menu_after_status_change(
|
||||
tmp_path: Path,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
settings = _settings(tmp_path)
|
||||
storage.init_health_db(settings.canonical_user_db_path())
|
||||
storage.init_state_db(settings)
|
||||
storage.set_user_menu_msg_id(settings, 123, 99)
|
||||
settings.canonical_user_status_path(123).write_text("{}", encoding="utf-8")
|
||||
|
||||
class FakeBot:
|
||||
edit_message_text = AsyncMock()
|
||||
|
||||
class FakeApp:
|
||||
bot = FakeBot()
|
||||
|
||||
monkeypatch.setattr(bot_app, "SETTINGS", settings)
|
||||
monkeypatch.setattr(bot_app, "ALLOWED_USER_ID", 123)
|
||||
monkeypatch.setattr(bot_app, "DB_PATH", str(settings.canonical_user_db_path()))
|
||||
monkeypatch.setattr(bot_app, "AUTO_MENU_REFRESH_INTERVAL", 0.01)
|
||||
|
||||
task = asyncio.create_task(bot_app.auto_refresh_main_menu_loop(FakeApp()))
|
||||
await asyncio.sleep(0.02)
|
||||
now = time.time() + 1
|
||||
path = settings.canonical_user_status_path(123)
|
||||
path.write_text('{"last_sync": 1}', encoding="utf-8")
|
||||
os.utime(path, (now, now))
|
||||
await asyncio.sleep(0.05)
|
||||
task.cancel()
|
||||
with pytest.raises(asyncio.CancelledError):
|
||||
await task
|
||||
|
||||
FakeApp.bot.edit_message_text.assert_awaited()
|
||||
@@ -1,76 +0,0 @@
|
||||
# SPDX-License-Identifier: GPL-3.0-or-later
|
||||
# Copyright (C) 2026 Alexey
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from miband_tracker.config import ConfigError, Settings, parse_user_ids
|
||||
|
||||
|
||||
def test_parse_user_ids_accepts_multiple_values() -> None:
|
||||
assert parse_user_ids("1,2", required=True) == [1, 2]
|
||||
assert parse_user_ids("1", required=True) == [1]
|
||||
|
||||
|
||||
def test_parse_user_ids_rejects_invalid_values() -> None:
|
||||
with pytest.raises(ConfigError):
|
||||
parse_user_ids("1,invalid", required=True)
|
||||
|
||||
|
||||
def test_settings_user_paths_prefer_existing_user_files(tmp_path: Path) -> None:
|
||||
user_id = 123
|
||||
(tmp_path / f"miband_{user_id}.db").write_text("", encoding="utf-8")
|
||||
(tmp_path / f"status_{user_id}.json").write_text("{}", encoding="utf-8")
|
||||
(tmp_path / f"token_{user_id}.json").write_text("{}", encoding="utf-8")
|
||||
|
||||
settings = Settings(
|
||||
data_dir=tmp_path,
|
||||
db_path=tmp_path / "miband.db",
|
||||
status_path=tmp_path / "status.json",
|
||||
bot_state_db_path=tmp_path / "fitness_bot_state.db",
|
||||
telegram_bot_token="token",
|
||||
telegram_allowed_user_ids=[user_id],
|
||||
sync_interval=900,
|
||||
query_duration=2,
|
||||
enable_fds_sleep_details=True,
|
||||
)
|
||||
|
||||
assert settings.telegram_allowed_user_id == user_id
|
||||
assert settings.user_db_path() == tmp_path / f"miband_{user_id}.db"
|
||||
assert settings.user_status_path() == tmp_path / f"status_{user_id}.json"
|
||||
assert settings.token_path() == tmp_path / f"token_{user_id}.json"
|
||||
|
||||
|
||||
def test_settings_falls_back_to_legacy_db_and_status(tmp_path: Path) -> None:
|
||||
settings = Settings(
|
||||
data_dir=tmp_path,
|
||||
db_path=tmp_path / "miband.db",
|
||||
status_path=tmp_path / "status.json",
|
||||
bot_state_db_path=tmp_path / "fitness_bot_state.db",
|
||||
telegram_bot_token="token",
|
||||
telegram_allowed_user_ids=[123],
|
||||
sync_interval=900,
|
||||
query_duration=2,
|
||||
enable_fds_sleep_details=True,
|
||||
)
|
||||
|
||||
assert settings.user_db_path() == tmp_path / "miband.db"
|
||||
assert settings.user_status_path() == tmp_path / "status.json"
|
||||
assert settings.canonical_user_db_path() == tmp_path / "miband_123.db"
|
||||
|
||||
|
||||
def test_settings_from_env_rejects_invalid_interval(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setenv("TELEGRAM_ALLOWED_USER_IDS", "123")
|
||||
monkeypatch.setenv("SYNC_INTERVAL", "soon")
|
||||
|
||||
with pytest.raises(ConfigError, match="SYNC_INTERVAL"):
|
||||
Settings.from_env()
|
||||
|
||||
|
||||
def test_settings_from_env_rejects_zero_query_duration(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setenv("TELEGRAM_ALLOWED_USER_IDS", "123")
|
||||
monkeypatch.setenv("QUERY_DURATION", "0")
|
||||
|
||||
with pytest.raises(ConfigError, match="QUERY_DURATION"):
|
||||
Settings.from_env()
|
||||
@@ -1,58 +0,0 @@
|
||||
# SPDX-License-Identifier: GPL-3.0-or-later
|
||||
# Copyright (C) 2026 Alexey
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from telegram import Update, User
|
||||
|
||||
from miband_tracker.bot import app
|
||||
from miband_tracker.config import Settings
|
||||
|
||||
|
||||
def test_dynamic_whitelisting(tmp_path: Path) -> None:
|
||||
# Setup temporary directories and config
|
||||
data_dir = tmp_path / "data"
|
||||
data_dir.mkdir()
|
||||
|
||||
# Override settings in app and os.environ
|
||||
os.environ["DATA_DIR"] = str(data_dir)
|
||||
os.environ["TELEGRAM_ALLOWED_USER_ID"] = ""
|
||||
|
||||
app.SETTINGS = Settings.from_env()
|
||||
app.ALLOWED_USER_ID = None
|
||||
|
||||
# Verify initial state
|
||||
assert app.ALLOWED_USER_ID is None
|
||||
assert not (data_dir / "allowed_user.id").exists()
|
||||
|
||||
# 1. First user tries to access
|
||||
user_1 = MagicMock(spec=User)
|
||||
user_1.id = 999999
|
||||
update_1 = MagicMock(spec=Update)
|
||||
update_1.effective_user = user_1
|
||||
|
||||
# First access should be allowed and should bind user_1
|
||||
assert app.is_allowed(update_1) is True
|
||||
assert app.ALLOWED_USER_ID == 999999
|
||||
assert (data_dir / "allowed_user.id").exists()
|
||||
assert (data_dir / "allowed_user.id").read_text(encoding="utf-8").strip() == "999999"
|
||||
|
||||
# 2. Second user tries to access
|
||||
user_2 = MagicMock(spec=User)
|
||||
user_2.id = 888888
|
||||
update_2 = MagicMock(spec=Update)
|
||||
update_2.effective_user = user_2
|
||||
|
||||
# Access for user_2 must be blocked
|
||||
assert app.is_allowed(update_2) is False
|
||||
|
||||
# 3. First user accesses again
|
||||
assert app.is_allowed(update_1) is True
|
||||
|
||||
# 4. Verify settings reloading
|
||||
reloaded_settings = Settings.from_env()
|
||||
assert reloaded_settings.telegram_allowed_user_id == 999999
|
||||
@@ -1,36 +0,0 @@
|
||||
# SPDX-License-Identifier: GPL-3.0-or-later
|
||||
# Copyright (C) 2026 Alexey
|
||||
|
||||
import struct
|
||||
|
||||
from miband_tracker.fds import parse_all_day_sleep_bytes
|
||||
|
||||
|
||||
def test_parse_all_day_sleep_bytes_reads_hr_and_spo2_records() -> None:
|
||||
blob = bytearray()
|
||||
blob += struct.pack("<I", 1_700_000_000)
|
||||
blob += bytes([12, 2, 0])
|
||||
blob += bytes([0b00000000, 0b11000000])
|
||||
blob += bytes([1])
|
||||
blob += struct.pack("<I", 1_700_000_100)
|
||||
blob += struct.pack("<I", 1_700_000_500)
|
||||
blob += bytes([80, 90])
|
||||
blob += struct.pack("<I", 120)
|
||||
blob += struct.pack("<I", 400)
|
||||
blob += struct.pack("<I", 1_700_000_090)
|
||||
blob += struct.pack("<I", 1_700_000_520)
|
||||
blob += struct.pack("<hhI", 60, 2, 1_700_000_100)
|
||||
blob += bytes([61, 62])
|
||||
blob += struct.pack("<hhI", 60, 2, 1_700_000_100)
|
||||
blob += bytes([97, 98])
|
||||
|
||||
parsed = parse_all_day_sleep_bytes(bytes(blob))
|
||||
|
||||
assert parsed is not None
|
||||
assert parsed["records"]["heart_rate"] == [(1_700_000_100, 61), (1_700_000_160, 62)]
|
||||
assert parsed["records"]["spo2"] == [(1_700_000_100, 97), (1_700_000_160, 98)]
|
||||
|
||||
|
||||
def test_parse_all_day_sleep_bytes_rejects_malformed_payloads() -> None:
|
||||
for blob in (b"", bytes(9), bytes(10), bytes(20)):
|
||||
assert parse_all_day_sleep_bytes(blob) is None
|
||||
@@ -1,38 +0,0 @@
|
||||
# SPDX-License-Identifier: GPL-3.0-or-later
|
||||
# Copyright (C) 2026 Alexey
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from miband_tracker.lock import LockUnavailable, exclusive_file_lock
|
||||
|
||||
|
||||
def test_exclusive_file_lock_rejects_second_process(tmp_path: Path) -> None:
|
||||
lock_path = tmp_path / "sync.lock"
|
||||
script = (
|
||||
"import sys, time\n"
|
||||
"from pathlib import Path\n"
|
||||
"from miband_tracker.lock import exclusive_file_lock\n"
|
||||
"with exclusive_file_lock(Path(sys.argv[1])):\n"
|
||||
" print('ready', flush=True)\n"
|
||||
" time.sleep(5)\n"
|
||||
)
|
||||
proc = subprocess.Popen(
|
||||
[sys.executable, "-c", script, str(lock_path)],
|
||||
stdout=subprocess.PIPE,
|
||||
text=True,
|
||||
)
|
||||
try:
|
||||
assert proc.stdout is not None
|
||||
assert proc.stdout.readline().strip() == "ready"
|
||||
with pytest.raises(LockUnavailable):
|
||||
with exclusive_file_lock(lock_path):
|
||||
pass
|
||||
finally:
|
||||
proc.terminate()
|
||||
proc.wait(timeout=5)
|
||||
@@ -1,35 +0,0 @@
|
||||
# SPDX-License-Identifier: GPL-3.0-or-later
|
||||
# Copyright (C) 2026 Alexey
|
||||
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
from mi_fitness.models import AuthToken
|
||||
|
||||
from miband_tracker.secure_files import save_auth_token, write_secret_json
|
||||
|
||||
|
||||
def test_write_secret_json_uses_private_file_mode(tmp_path: Path) -> None:
|
||||
path = tmp_path / "token.json"
|
||||
|
||||
write_secret_json(path, {"service_token": "secret"})
|
||||
|
||||
assert json.loads(path.read_text(encoding="utf-8")) == {"service_token": "secret"}
|
||||
assert stat_mode(path) == 0o600
|
||||
|
||||
|
||||
def test_save_auth_token_preserves_target_relative_uid(tmp_path: Path) -> None:
|
||||
path = tmp_path / "token.json"
|
||||
write_secret_json(path, {"target_relative_uid": "42"})
|
||||
|
||||
save_auth_token(AuthToken(user_id="11", service_token="svc", ssecurity="sec"), path)
|
||||
|
||||
data = json.loads(path.read_text(encoding="utf-8"))
|
||||
assert data["user_id"] == "11"
|
||||
assert data["target_relative_uid"] == "42"
|
||||
assert stat_mode(path) == 0o600
|
||||
|
||||
|
||||
def stat_mode(path: Path) -> int:
|
||||
return os.stat(path).st_mode & 0o777
|
||||
@@ -1,19 +0,0 @@
|
||||
# SPDX-License-Identifier: GPL-3.0-or-later
|
||||
# Copyright (C) 2026 Alexey
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import io
|
||||
|
||||
from miband_tracker.stdio import safe_print
|
||||
|
||||
|
||||
def test_safe_print_does_not_crash_on_non_ascii_with_charmap_stream() -> None:
|
||||
raw = io.BytesIO()
|
||||
stream = io.TextIOWrapper(raw, encoding="cp1251", errors="strict")
|
||||
|
||||
safe_print("API 业务错误", file=stream, flush=True)
|
||||
stream.flush()
|
||||
|
||||
assert b"API " in raw.getvalue()
|
||||
assert b"\\u4e1a" in raw.getvalue()
|
||||
@@ -1,114 +0,0 @@
|
||||
# SPDX-License-Identifier: GPL-3.0-or-later
|
||||
# Copyright (C) 2026 Alexey
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from miband_tracker import storage
|
||||
from miband_tracker.config import Settings
|
||||
|
||||
|
||||
def _settings(tmp_path: Path) -> Settings:
|
||||
return Settings(
|
||||
data_dir=tmp_path,
|
||||
db_path=tmp_path / "miband.db",
|
||||
status_path=tmp_path / "status.json",
|
||||
bot_state_db_path=tmp_path / "fitness_bot_state.db",
|
||||
telegram_bot_token="token",
|
||||
telegram_allowed_user_id=123,
|
||||
sync_interval=900,
|
||||
query_duration=2,
|
||||
enable_fds_sleep_details=True,
|
||||
)
|
||||
|
||||
|
||||
def test_init_health_db_is_idempotent(tmp_path: Path) -> None:
|
||||
db_path = tmp_path / "miband_123.db"
|
||||
|
||||
storage.init_health_db(db_path)
|
||||
storage.init_health_db(db_path)
|
||||
|
||||
with storage.sqlite_conn(db_path) as conn:
|
||||
tables = {
|
||||
row["name"]
|
||||
for row in conn.execute("SELECT name FROM sqlite_master WHERE type = 'table'").fetchall()
|
||||
}
|
||||
assert {
|
||||
"steps_daily",
|
||||
"sleep_daily",
|
||||
"heart_rate",
|
||||
"blood_oxygen",
|
||||
"stress",
|
||||
"calories_daily",
|
||||
"weight",
|
||||
"workouts",
|
||||
}.issubset(tables)
|
||||
|
||||
|
||||
def test_extended_health_tables_have_expected_columns(tmp_path: Path) -> None:
|
||||
db_path = tmp_path / "miband_123.db"
|
||||
storage.init_health_db(db_path)
|
||||
|
||||
with storage.sqlite_conn(db_path, row_factory=False) as conn:
|
||||
columns = {
|
||||
table: {
|
||||
row[1]
|
||||
for row in conn.execute(f"PRAGMA table_info({table})").fetchall()
|
||||
}
|
||||
for table in ("stress", "calories_daily", "weight", "workouts")
|
||||
}
|
||||
|
||||
assert columns["stress"] == {"timestamp", "value"}
|
||||
assert {
|
||||
"date",
|
||||
"total_cal",
|
||||
"active_cal",
|
||||
"valid_stand_hours",
|
||||
"intensity_minutes",
|
||||
"last_sync",
|
||||
}.issubset(columns["calories_daily"])
|
||||
assert {"timestamp", "weight_kg", "bmi", "body_fat_pct"}.issubset(columns["weight"])
|
||||
assert {
|
||||
"workout_id",
|
||||
"sport_type",
|
||||
"start_time",
|
||||
"end_time",
|
||||
"duration_sec",
|
||||
"calories",
|
||||
"avg_hr",
|
||||
"max_hr",
|
||||
"min_hr",
|
||||
"watermark",
|
||||
"raw_json",
|
||||
}.issubset(columns["workouts"])
|
||||
|
||||
|
||||
def test_zip_export_includes_non_empty_tables_only(tmp_path: Path) -> None:
|
||||
settings = _settings(tmp_path)
|
||||
db_path = settings.canonical_user_db_path()
|
||||
storage.init_health_db(db_path)
|
||||
with storage.sqlite_conn(db_path, row_factory=False) as conn:
|
||||
conn.execute(
|
||||
"INSERT INTO steps_daily (date, total_steps, calories, distance_m, last_sync) VALUES (?, ?, ?, ?, ?)",
|
||||
("2026-05-24", 1000, 10.0, 800.0, 1),
|
||||
)
|
||||
conn.execute(
|
||||
"INSERT INTO stress (timestamp, value) VALUES (?, ?)",
|
||||
(1779760000, 42),
|
||||
)
|
||||
conn.execute(
|
||||
"""
|
||||
INSERT INTO workouts
|
||||
(workout_id, sport_type, start_time, end_time, duration_sec,
|
||||
calories, avg_hr, max_hr, min_hr, watermark, raw_json)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
""",
|
||||
("w1", "free_training", 1779760000, 1779760600, 600, 55.0, 110, 130, 90, 1, "{}"),
|
||||
)
|
||||
conn.commit()
|
||||
|
||||
archive = storage.zip_export(settings)
|
||||
|
||||
assert archive.getbuffer().nbytes > 0
|
||||
assert b"steps_daily.csv" in archive.getvalue()
|
||||
assert b"stress.csv" in archive.getvalue()
|
||||
assert b"workouts.csv" in archive.getvalue()
|
||||
@@ -1,77 +0,0 @@
|
||||
# SPDX-License-Identifier: GPL-3.0-or-later
|
||||
# Copyright (C) 2026 Alexey
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from miband_tracker.config import Settings
|
||||
from miband_tracker.storage import init_health_db, sqlite_conn
|
||||
from miband_tracker.sync import _sync_workouts, run_sync
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_run_sync_missing_token_returns_failed_result(tmp_path: Path) -> None:
|
||||
settings = Settings(
|
||||
data_dir=tmp_path,
|
||||
db_path=tmp_path / "miband.db",
|
||||
status_path=tmp_path / "status.json",
|
||||
bot_state_db_path=tmp_path / "fitness_bot_state.db",
|
||||
telegram_bot_token="token",
|
||||
telegram_allowed_user_id=123,
|
||||
sync_interval=0,
|
||||
query_duration=2,
|
||||
enable_fds_sleep_details=True,
|
||||
)
|
||||
|
||||
result = await run_sync(settings=settings)
|
||||
|
||||
assert not result.success
|
||||
assert result.user_id == 123
|
||||
assert "Token file not found" in (result.error or "")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_sync_workouts_inserts_watermark_records(tmp_path: Path) -> None:
|
||||
db_path = tmp_path / "miband_123.db"
|
||||
init_health_db(db_path)
|
||||
|
||||
class FakeClient:
|
||||
async def _request(self, method, path, params):
|
||||
assert method == "GET"
|
||||
assert path == "/app/v1/data/get_sport_records_by_watermark"
|
||||
assert params["relative_uid"] == 456
|
||||
return {
|
||||
"result": {
|
||||
"has_more": False,
|
||||
"sport_records": [
|
||||
{
|
||||
"sid": "w1",
|
||||
"key": "free_training",
|
||||
"time": 1779752903,
|
||||
"watermark": 12345,
|
||||
"value": (
|
||||
'{"start_time": 1779752903, "end_time": 1779753050, '
|
||||
'"duration": 142, "calories": 7, "avg_hrm": 95, '
|
||||
'"max_hrm": 152, "min_hrm": 80}'
|
||||
),
|
||||
}
|
||||
],
|
||||
}
|
||||
}
|
||||
|
||||
counters = {"workouts": 0}
|
||||
with sqlite_conn(db_path, row_factory=False) as conn:
|
||||
cursor = conn.cursor()
|
||||
await _sync_workouts(FakeClient(), cursor, counters, 456)
|
||||
conn.commit()
|
||||
|
||||
with sqlite_conn(db_path) as conn:
|
||||
row = conn.execute("SELECT * FROM workouts WHERE workout_id = ?", ("w1",)).fetchone()
|
||||
|
||||
assert counters["workouts"] == 1
|
||||
assert row is not None
|
||||
assert row["sport_type"] == "free_training"
|
||||
assert row["duration_sec"] == 142
|
||||
assert row["avg_hr"] == 95
|
||||
assert row["watermark"] == 12345
|
||||
@@ -0,0 +1,18 @@
|
||||
import { describe, expect, test } from "bun:test";
|
||||
import { buildEncryptedParams, computeSignedNonce, decryptData, encryptData } from "../src/xiaomi/client.js";
|
||||
|
||||
describe("Xiaomi crypto", () => {
|
||||
test("round-trips encrypted payloads and request params", () => {
|
||||
const ssecurity = Buffer.from("test-ssecurity-key").toString("base64");
|
||||
const nonce = Buffer.from("123456789012").toString("base64");
|
||||
const signed = computeSignedNonce(ssecurity, nonce);
|
||||
expect(decryptData(signed, encryptData(signed, '{"中文":"тест"}'))).toBe('{"中文":"тест"}');
|
||||
const params = buildEncryptedParams("POST", "/test", ssecurity, { test: "hello", num: 42 });
|
||||
expect(JSON.parse(decryptData(computeSignedNonce(ssecurity, params._nonce ?? ""), params.data ?? ""))).toEqual({
|
||||
test: "hello",
|
||||
num: 42,
|
||||
});
|
||||
expect(params.signature).toBeString();
|
||||
expect(params.rc4_hash__).toBeString();
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user