solduri2roa/tests/test_citire_fdb.py

"""Teste pentru citire_fdb: balanta din Firebird identica cu exportul .xlsx.

Poarta: `balanta` produsa din copia unei baze `CONT_BAZA.FDB` trebuie sa fie
identica (pana la octet) cu cea produsa de `citire_xlsx.py` din exportul de foaie
de calcul, pe firma de joaca, luna 09/2026. Baza originala NU se deschide: se
copiaza intr-un director temporar.

Pe o masina fara SAGA (fara fbclient.dll, baza, driver sau export) testele se
sar, nu esueaza.

Rulare:  py -m unittest discover -s tests -v
"""
import os
import subprocess
import pathlib
import shutil
import sys
import tempfile
import unittest

RADACINA = pathlib.Path(__file__).resolve().parent.parent
if str(RADACINA) not in sys.path:
    sys.path.insert(0, str(RADACINA))

import citire_fdb
import citire_xlsx

FBCLIENT = pathlib.Path(r"C:\Program Files\Firebird\Firebird30_Saga\fbclient.dll")
BAZA = pathlib.Path(r"D:\SAGA250909\0001\CONT_BAZA.FDB")
XLSX = RADACINA / "exemple" / "saga-sqlite-balanta-09-2026.xlsx"

AN, LUNA = 2026, 9
TOL = 0.01


def _driver_disponibil():
    try:
        import firebird.driver  # noqa: F401
        return True
    except Exception:
        return False


class TestCitireFdb(unittest.TestCase):

    @classmethod
    def setUpClass(cls):
        if not FBCLIENT.exists():
            raise unittest.SkipTest("lipseste fbclient.dll: %s" % FBCLIENT)
        if not BAZA.exists():
            raise unittest.SkipTest("lipseste baza Firebird: %s" % BAZA)
        if not _driver_disponibil():
            raise unittest.SkipTest("lipseste pachetul firebird-driver")
        cls._tmp = tempfile.TemporaryDirectory(prefix="citire_fdb_")
        cls.baza = os.path.join(cls._tmp.name, "CONT_BAZA.FDB")
        shutil.copy2(str(BAZA), cls.baza)
        cls.dir = cls._tmp.name

    @classmethod
    def tearDownClass(cls):
        cls._tmp.cleanup()

    def test_poarta_identica_cu_xlsx(self):
        if not XLSX.exists():
            self.skipTest("lipseste exportul %s" % XLSX)
        rows_xlsx = citire_xlsx.citeste_balanta(str(XLSX))
        # exportul firmei de joaca e fara analitice; cititorul le emite si pe ele
        rows_fdb = [r for r in citire_fdb.citeste_balanta(self.baza, AN, LUNA)
                    if "." not in r["cont"]]
        self.assertTrue(rows_xlsx, "exportul xlsx nu are randuri")
        self.assertEqual(rows_xlsx, rows_fdb,
                         "balanta din FDB difera de exportul xlsx")

        # comparatie stricta: acelasi fisier CSV, la octet
        cale_xlsx = os.path.join(self.dir, "din_xlsx.csv")
        cale_fdb = os.path.join(self.dir, "din_fdb.csv")
        citire_xlsx.scrie_csv(rows_xlsx, cale_xlsx)
        citire_fdb.scrie_csv(rows_fdb, cale_fdb, citire_fdb.COLOANE_BALANTA)
        with open(cale_xlsx, "rb") as f:
            a = f.read()
        with open(cale_fdb, "rb") as f:
            b = f.read()
        self.assertEqual(a, b, "CSV-urile difera la nivel de octet")

    def test_sum_registru_egal_netul_balantei(self):
        d_an, d_luna, d_urm = citire_fdb._limite(AN, LUNA)
        con = citire_fdb._conecteaza(self.baza)
        try:
            deb, cred = citire_fdb._select(
                con,
                "SELECT "
                "SUM(CASE WHEN TRIM(CONT_D) NOT IN ('', '%') THEN SUMA ELSE 0 END), "
                "SUM(CASE WHEN TRIM(CONT_C) NOT IN ('', '%') THEN SUMA ELSE 0 END) "
                "FROM REGISTRU WHERE DATA < ?", (d_urm,))[0]
        finally:
            con.rollback()
            con.close()
        self.assertAlmostEqual(float(deb), float(cred), delta=TOL,
                               msg="SUM(REGISTRU) nu e echilibrat pe laturi")

        rows = citire_fdb.citeste_balanta(self.baza, AN, LUNA)
        net_balanta = sum(r["total_d"] - r["total_c"] for r in rows
                          if "." not in r["cont"])
        self.assertAlmostEqual(float(deb) - float(cred), net_balanta, delta=TOL,
                               msg="netul din REGISTRU != netul balantei")

    def test_scadent_null_cade_pe_ultima_zi_a_lunii(self):
        rows = citire_fdb.citeste_facturi(self.baza, AN, LUNA)
        self.assertTrue(rows, "nu s-au citit facturi")
        asteptat = citire_fdb._ultima_zi(AN, LUNA).isoformat()
        # singura factura din firma de joaca are SCADENT NULL in INTRD
        self.assertEqual(rows[0]["scadent"], asteptat)

    def test_fbk_identic_cu_fdb(self):
        """Backupul gbak (.fbk) se restaureaza in temporar si da aceleasi fisiere."""
        fbk = os.path.join(self.dir, "firma.fbk")
        subprocess.run([citire_fdb.GBAK, "-b", self.baza, fbk, "-user",
                        citire_fdb.UTILIZATOR, "-password", citire_fdb.PAROLA], check=True)
        a = citire_fdb.produce(self.baza, "A", AN, LUNA, os.path.join(self.dir, "a"))
        b = citire_fdb.produce(fbk, "B", AN, LUNA, os.path.join(self.dir, "b"))
        self.assertEqual(a, b)

    def test_cont_colector_exclus(self):
        rows = citire_fdb.citeste_balanta(self.baza, AN, LUNA)
        self.assertNotIn("%", [r["cont"] for r in rows])
        self.assertTrue(all(r["cont"] for r in rows), "rand fara cont")

    def test_parteneri_si_facturi(self):
        parteneri = citire_fdb.citeste_parteneri(self.baza)
        self.assertTrue(parteneri, "nu s-au citit parteneri")
        for r in parteneri:
            self.assertTrue(r["cont_analitic"])
            self.assertTrue(r["cod_fiscal"],
                            "cod fiscal gol pentru %s" % r["cont_analitic"])
        # Codul fiscal lipsa in SAGA cade pe contul analitic (citire_fdb.py). Pe firma
        # de joaca ROMFAST (401.00001) nu are COD_FISCAL, TRANSPORT (401.00002) are CUI
        # real. Asta dovedeste ca drumul FDB NU scuteste de etapa 5b (cautarea ANAF).
        provizorii = {r["cont_analitic"] for r in parteneri
                      if r["cod_fiscal"] == r["cont_analitic"]}
        self.assertEqual(provizorii, {"401.00001"},
                         "s-a schimbat setul de parteneri fara CUI real")

        facturi = citire_fdb.citeste_facturi(self.baza, AN, LUNA)
        for r in facturi:
            self.assertTrue(r["cont_analitic"])
            self.assertTrue(r["scadent"])


if __name__ == "__main__":
    unittest.main()