test_insert_workload.py - master - AxeOn/ProjectAi

test_insert_workload.py
102 строки3 KB
Дмитрий
chore: base state before refactoring (office_planner_v5)
25 апр 2026, 15:01
25 апр 2026, 15:01bc896b2
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Тестовая запись напрямую в workload
"""

import sqlite3
import os

DB_PATH = "office_planner.db"

print("=" * 80)
print("ТЕСТОВАЯ ЗАПИСЬ В WORKLOAD")
print("=" * 80)

# Проверяем путь к БД
print(f"\n1. Путь к БД: {os.path.abspath(DB_PATH)}")
print(f"   Файл существует: {os.path.exists(DB_PATH)}")

if not os.path.exists(DB_PATH):
    print("\n❌ ОШИБКА: Файл БД не найден!")
    print("   Укажите правильный путь в скрипте.")
    exit(1)

# Подключаемся
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()

# Проверяем таблицу
cursor.execute("""
    SELECT name FROM sqlite_master 
    WHERE type='table' AND name='workload'
""")

if not cursor.fetchone():
    print("\n❌ Таблица workload не существует!")
    exit(1)

print("\n2. Таблица workload: ✅ существует")

# Считаем записи ДО
cursor.execute("SELECT COUNT(*) FROM workload WHERE source='monthly'")
before = cursor.fetchone()[0]
print(f"\n3. Записей ДО вставки: {before}")

# Вставляем ТЕСТОВУЮ запись
print("\n4. Вставляем тестовую запись...")
try:
    cursor.execute("""
        INSERT INTO workload 
        (period, emp_name, service_name, count, price, total_sum, created_at, source)
        VALUES (?, ?, ?, ?, ?, ?, ?, ?)
    """, ('2026-01', 'ТЕСТ Тестовый Тестович', 'ТЕСТ Услуга', 999, 0, 0, '2026-01-01', 'monthly'))
    
    conn.commit()
    print("   ✅ INSERT выполнен")
    print("   ✅ COMMIT выполнен")
    
    # WAL checkpoint
    cursor.execute("PRAGMA wal_checkpoint(FULL)")
    print("   ✅ WAL checkpoint выполнен")
    
except Exception as e:
    print(f"   ❌ ОШИБКА: {e}")
    conn.rollback()
    exit(1)

# Считаем записи ПОСЛЕ
cursor.execute("SELECT COUNT(*) FROM workload WHERE source='monthly'")
after = cursor.fetchone()[0]
print(f"\n5. Записей ПОСЛЕ вставки: {after}")

if after > before:
    print(f"\n✅ УСПЕХ! Добавлено {after - before} записей")
    
    # Показываем тестовую запись
    cursor.execute("""
        SELECT emp_name, service_name, count, period
        FROM workload
        WHERE emp_name LIKE 'ТЕСТ%'
        LIMIT 1
    """)
    row = cursor.fetchone()
    if row:
        print(f"\nТестовая запись:")
        print(f"  Сотрудник: {row[0]}")
        print(f"  Услуга: {row[1]}")
        print(f"  Количество: {row[2]}")
        print(f"  Период: {row[3]}")
else:
    print(f"\n❌ ПРОВАЛ! Запись НЕ СОХРАНИЛАСЬ!")
    print("\nВозможные причины:")
    print("  1. Транзакция откатилась")
    print("  2. Используется другой файл БД")
    print("  3. Проблема с WAL режимом")

conn.close()

print("\n" + "=" * 80)
print("Теперь запустите check_monthly_data.py и проверьте")
print("появилась ли тестовая запись 'ТЕСТ Тестовый Тестович'")
print("=" * 80)