214 lines
7.0 KiB
Python
214 lines
7.0 KiB
Python
|
# -*- coding: utf-8 -*-
|
||
|
|
"""Tests für practice_id-Sync (Textblöcke, Korrekturen, Autotext) und renewal_date_de."""
|
||
|
|
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
import sqlite3
|
||
|
|
import sys
|
||
|
|
import tempfile
|
||
|
|
import unittest
|
||
|
|
from pathlib import Path
|
||
|
|
from unittest.mock import MagicMock
|
||
|
|
|
||
|
|
ROOT = Path(__file__).resolve().parents[1]
|
||
|
|
if str(ROOT) not in sys.path:
|
||
|
|
sys.path.insert(0, str(ROOT))
|
||
|
|
|
||
|
|
from aza_sync_items import (
|
||
|
|
ensure_sync_items_schema,
|
||
|
|
items_to_local_textblocks,
|
||
|
|
list_sync_items,
|
||
|
|
local_autotext_to_items,
|
||
|
|
local_korrekturen_to_items,
|
||
|
|
local_textblocks_to_items,
|
||
|
|
soft_delete_sync_item,
|
||
|
|
upsert_sync_item,
|
||
|
|
)
|
||
|
|
from aza_ai_budget import budget_json_for_client
|
||
|
|
|
||
|
|
|
||
|
|
class TestSyncItemsSchema(unittest.TestCase):
|
||
|
|
def test_schema_idempotent(self):
|
||
|
|
td = tempfile.mkdtemp()
|
||
|
|
try:
|
||
|
|
db = Path(td) / "t.sqlite"
|
||
|
|
con = sqlite3.connect(db)
|
||
|
|
try:
|
||
|
|
ensure_sync_items_schema(con)
|
||
|
|
ensure_sync_items_schema(con)
|
||
|
|
cols = {
|
||
|
|
r[1]
|
||
|
|
for r in con.execute("PRAGMA table_info(synced_user_items)").fetchall()
|
||
|
|
}
|
||
|
|
finally:
|
||
|
|
con.close()
|
||
|
|
self.assertIn("practice_id", cols)
|
||
|
|
self.assertIn("item_type", cols)
|
||
|
|
finally:
|
||
|
|
try:
|
||
|
|
os.remove(Path(td) / "t.sqlite")
|
||
|
|
except OSError:
|
||
|
|
pass
|
||
|
|
os.rmdir(td)
|
||
|
|
|
||
|
|
|
||
|
|
class TestSyncItemsDb(unittest.TestCase):
|
||
|
|
def tearDown(self):
|
||
|
|
import shutil
|
||
|
|
|
||
|
|
try:
|
||
|
|
shutil.rmtree(self.td, ignore_errors=True)
|
||
|
|
except Exception:
|
||
|
|
pass
|
||
|
|
|
||
|
|
def setUp(self):
|
||
|
|
self.td = tempfile.mkdtemp()
|
||
|
|
self.db = Path(self.td) / "stripe_webhook.sqlite"
|
||
|
|
with sqlite3.connect(self.db) as con:
|
||
|
|
ensure_sync_items_schema(con)
|
||
|
|
con.execute(
|
||
|
|
"""
|
||
|
|
CREATE TABLE licenses (
|
||
|
|
subscription_id TEXT PRIMARY KEY,
|
||
|
|
customer_id TEXT,
|
||
|
|
status TEXT,
|
||
|
|
lookup_key TEXT,
|
||
|
|
allowed_users INTEGER,
|
||
|
|
devices_per_user INTEGER,
|
||
|
|
customer_email TEXT,
|
||
|
|
client_reference_id TEXT,
|
||
|
|
current_period_end INTEGER,
|
||
|
|
updated_at INTEGER NOT NULL,
|
||
|
|
practice_id TEXT,
|
||
|
|
current_period_start INTEGER
|
||
|
|
)
|
||
|
|
"""
|
||
|
|
)
|
||
|
|
con.execute(
|
||
|
|
"""
|
||
|
|
INSERT INTO licenses (
|
||
|
|
subscription_id, customer_id, status, lookup_key,
|
||
|
|
allowed_users, devices_per_user, customer_email,
|
||
|
|
client_reference_id, current_period_end, updated_at,
|
||
|
|
practice_id, current_period_start
|
||
|
|
) VALUES ('sub_a', 'c1', 'active', 'k1', 5, 2, 'a@test.ch',
|
||
|
|
'', 2000000000, 1, 'prac_alpha', 1900000000)
|
||
|
|
"""
|
||
|
|
)
|
||
|
|
con.execute(
|
||
|
|
"""
|
||
|
|
INSERT INTO licenses (
|
||
|
|
subscription_id, customer_id, status, lookup_key,
|
||
|
|
allowed_users, devices_per_user, customer_email,
|
||
|
|
client_reference_id, current_period_end, updated_at,
|
||
|
|
practice_id, current_period_start
|
||
|
|
) VALUES ('sub_b', 'c2', 'active', 'k2', 5, 2, 'b@test.ch',
|
||
|
|
'', 2000000000, 1, 'prac_beta', 1900000000)
|
||
|
|
"""
|
||
|
|
)
|
||
|
|
con.commit()
|
||
|
|
|
||
|
|
def test_practice_isolation_and_soft_delete(self):
|
||
|
|
with sqlite3.connect(self.db) as con:
|
||
|
|
upsert_sync_item(
|
||
|
|
con,
|
||
|
|
"prac_alpha",
|
||
|
|
{
|
||
|
|
"id": "tb_1",
|
||
|
|
"item_type": "textblock",
|
||
|
|
"title": "A",
|
||
|
|
"content": "alpha",
|
||
|
|
"sort_order": 1,
|
||
|
|
},
|
||
|
|
)
|
||
|
|
upsert_sync_item(
|
||
|
|
con,
|
||
|
|
"prac_beta",
|
||
|
|
{
|
||
|
|
"id": "tb_beta_1",
|
||
|
|
"item_type": "textblock",
|
||
|
|
"title": "B",
|
||
|
|
"content": "beta",
|
||
|
|
"sort_order": 1,
|
||
|
|
},
|
||
|
|
)
|
||
|
|
a_items = list_sync_items(con, "prac_alpha", "textblock")
|
||
|
|
b_items = list_sync_items(con, "prac_beta", "textblock")
|
||
|
|
self.assertEqual(len(a_items), 1)
|
||
|
|
self.assertEqual(a_items[0]["content"], "alpha")
|
||
|
|
self.assertEqual(b_items[0]["content"], "beta")
|
||
|
|
soft_delete_sync_item(con, "prac_alpha", "tb_1")
|
||
|
|
a2 = list_sync_items(con, "prac_alpha", "textblock")
|
||
|
|
self.assertEqual(len(a2), 0)
|
||
|
|
|
||
|
|
def test_item_types_separate(self):
|
||
|
|
with sqlite3.connect(self.db) as con:
|
||
|
|
upsert_sync_item(
|
||
|
|
con,
|
||
|
|
"prac_alpha",
|
||
|
|
{
|
||
|
|
"id": "at_x",
|
||
|
|
"item_type": "autotext",
|
||
|
|
"trigger": "kg",
|
||
|
|
"content": "Krankengeschichte",
|
||
|
|
"sort_order": 1,
|
||
|
|
},
|
||
|
|
)
|
||
|
|
upsert_sync_item(
|
||
|
|
con,
|
||
|
|
"prac_alpha",
|
||
|
|
{
|
||
|
|
"id": "corr_1",
|
||
|
|
"item_type": "correction",
|
||
|
|
"title": "medikamente",
|
||
|
|
"trigger": "asprin",
|
||
|
|
"content": "Aspirin",
|
||
|
|
"sort_order": 1,
|
||
|
|
},
|
||
|
|
)
|
||
|
|
at = list_sync_items(con, "prac_alpha", "autotext")
|
||
|
|
co = list_sync_items(con, "prac_alpha", "correction")
|
||
|
|
self.assertEqual(len(at), 1)
|
||
|
|
self.assertEqual(len(co), 1)
|
||
|
|
|
||
|
|
|
||
|
|
class TestLocalMapping(unittest.TestCase):
|
||
|
|
def test_textblock_roundtrip(self):
|
||
|
|
local = {
|
||
|
|
"1": {"name": "Urtikaria", "content": "Haut", "updated_at": "t"},
|
||
|
|
"2": {"name": "TB2", "content": "", "updated_at": "t"},
|
||
|
|
}
|
||
|
|
items = local_textblocks_to_items(local)
|
||
|
|
back = items_to_local_textblocks(items)
|
||
|
|
self.assertEqual(back["1"]["name"], "Urtikaria")
|
||
|
|
self.assertGreaterEqual(len(back), 2)
|
||
|
|
|
||
|
|
def test_korrekturen_mapping(self):
|
||
|
|
k = {"medikamente": {"asprin": "Aspirin"}}
|
||
|
|
items = local_korrekturen_to_items(k)
|
||
|
|
self.assertEqual(items[0]["item_type"], "correction")
|
||
|
|
|
||
|
|
def test_autotext_mapping(self):
|
||
|
|
at = {"entries": {"kg": "Krankengeschichte"}}
|
||
|
|
items = local_autotext_to_items(at)
|
||
|
|
self.assertEqual(items[0]["trigger"], "kg")
|
||
|
|
|
||
|
|
|
||
|
|
class TestBudgetRenewalField(unittest.TestCase):
|
||
|
|
def test_renewal_date_de_in_client_json(self):
|
||
|
|
snap = {
|
||
|
|
"ok": True,
|
||
|
|
"active": True,
|
||
|
|
"available_percent": 76,
|
||
|
|
"period_end": 1781817600,
|
||
|
|
"show_warning": False,
|
||
|
|
"user_label": "KI-Kontingent: 76 % verfügbar",
|
||
|
|
}
|
||
|
|
body = budget_json_for_client(snap)
|
||
|
|
self.assertIn("renewal_date_de", body)
|
||
|
|
self.assertTrue(body["renewal_date_de"])
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
unittest.main()
|