gru/ksef_app
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

push

Browse Source
This commit is contained in:
Mateusz Gruszczyński
2026-03-13 11:03:13 +01:00
commit 35571df778
132 changed files with 11197 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

39
tests/conftest.py Normal file
View File

@@ -0,0 +1,39 @@
import pytest
from werkzeug.security import generate_password_hash
from app import create_app
from config import TestConfig
from app.extensions import db
from app.models.user import User
from app.models.company import Company, UserCompanyAccess
from app.models.setting import AppSetting
@pytest.fixture()
def app():
app = create_app(TestConfig)
with app.app_context():
db.create_all()
company = Company(name='Test Co', tax_id='123', sync_enabled=True, sync_interval_minutes=60)
db.session.add(company)
db.session.flush()
user = User(email='admin@example.com', name='Admin', password_hash=generate_password_hash('Admin123!'), role='admin')
db.session.add(user)
db.session.flush()
db.session.add(UserCompanyAccess(user_id=user.id, company_id=company.id, access_level='full'))
AppSetting.set(f'company.{company.id}.notify.enabled', 'true')
AppSetting.set(f'company.{company.id}.notify.min_amount', '0')
AppSetting.set(f'company.{company.id}.ksef.mock_mode', 'true')
db.session.commit()
yield app
db.drop_all()
@pytest.fixture()
def client(app):
return app.test_client()
@pytest.fixture()
def auth_client(client):
client.post('/auth/login', data={'email': 'admin@example.com', 'password': 'Admin123!'}, follow_redirects=True)
return client

22
tests/test_admin_nav.py Normal file
View File

@@ -0,0 +1,22 @@
def test_admin_nav_visible_on_users(auth_client):
response = auth_client.get('/admin/users')
body = response.get_data(as_text=True)
assert response.status_code == 200
assert 'Logi audytu' in body
assert 'Firmy' in body
def test_admin_nav_visible_on_companies(auth_client):
response = auth_client.get('/admin/companies')
body = response.get_data(as_text=True)
assert response.status_code == 200
assert 'Użytkownicy' in body
assert 'Logi audytu' in body
def test_admin_nav_visible_on_audit(auth_client):
response = auth_client.get('/admin/audit')
body = response.get_data(as_text=True)
assert response.status_code == 200
assert 'Użytkownicy' in body
assert 'Firmy' in body

187
tests/test_admin_system.py Normal file
View File

@@ -0,0 +1,187 @@
from datetime import datetime, timedelta
from pathlib import Path
from requests import exceptions as requests_exceptions
from app.extensions import db
from app.models.audit_log import AuditLog
from app.models.company import Company
from app.models.invoice import MailDelivery, NotificationLog, SyncEvent
from app.models.setting import AppSetting
from app.models.sync_log import SyncLog
from app.services.ksef_service import KSeFService, RequestsKSeFAdapter
class DummyResponse:
def __init__(self, payload, status_code=200):
self._payload = payload
self.status_code = status_code
self.content = b'{}'
self.text = '{}'
def json(self):
return self._payload
def test_admin_system_data_page(auth_client, monkeypatch):
from app.services.ceidg_service import CeidgService
monkeypatch.setattr(CeidgService, 'diagnostics', lambda self: {
'status': 'ok',
'message': 'HTTP 200',
'environment': 'test',
'url': 'https://test.example/ceidg',
'sample': {'ok': True},
'technical_details': None,
})
response = auth_client.get('/admin/system-data')
body = response.get_data(as_text=True)
assert response.status_code == 200
assert 'Dane systemowe' in body
assert 'Połączenie KSeF' in body
assert 'Połączenie CEIDG' in body
assert 'Proces i health systemu' in body
assert 'Użytkownicy' in body
assert 'Diagnoza' not in body
assert 'Co sprawdzić' not in body
def test_admin_health_redirects_to_system_data(auth_client):
response = auth_client.get('/admin/health', follow_redirects=False)
assert response.status_code == 302
assert '/admin/system-data' in response.headers['Location']
def test_ksef_diagnostics_mock(app):
with app.app_context():
company = Company.query.first()
data = KSeFService(company_id=company.id).diagnostics()
assert data['status'] == 'mock'
assert data['base_url'] == 'mock://ksef'
assert 'documentExample' in data['sample']
def test_ksef_diagnostics_requests_adapter(app, monkeypatch):
with app.app_context():
company = Company.query.first()
AppSetting.set(f'company.{company.id}.ksef.mock_mode', 'false')
db.session.commit()
monkeypatch.setattr(RequestsKSeFAdapter, '_request', lambda self, method, path, params=None, json=None: {'status': 'healthy', 'version': 'demo'})
data = KSeFService(company_id=company.id).diagnostics()
assert data['status'] == 'ok'
assert data['sample']['status'] == 'healthy'
assert data['base_url'].startswith('https://')
def test_ceidg_diagnostics_timeout(app, monkeypatch):
from app.services.ceidg_service import CeidgService
def raise_timeout(*args, **kwargs):
raise requests_exceptions.ConnectTimeout('connect timeout')
monkeypatch.setattr('app.services.ceidg_service.requests.get', raise_timeout)
with app.app_context():
AppSetting.set('ceidg.environment', 'test')
AppSetting.set('ceidg.api_key', 'diagnostic-key', encrypt=True)
db.session.commit()
data = CeidgService().diagnostics()
assert data['status'] == 'error'
assert 'Timeout' in data['message']
assert data['environment'] == 'test'
def test_cleanup_logs_removes_old_records(app, auth_client):
with app.app_context():
old_dt = datetime.utcnow() - timedelta(days=120)
company = Company.query.first()
invoice = company.invoices.first()
if invoice is None:
from app.models.invoice import Invoice, InvoiceType, InvoiceStatus
invoice = Invoice(company_id=company.id, ksef_number='X1', invoice_number='FV/1', contractor_name='Test', issue_date=old_dt.date(), net_amount=1, vat_amount=0.23, gross_amount=1.23, invoice_type=InvoiceType.SALE, status=InvoiceStatus.NEW)
db.session.add(invoice)
db.session.flush()
db.session.add(AuditLog(action='old', target_type='system', created_at=old_dt, updated_at=old_dt))
db.session.add(SyncLog(sync_type='manual', status='done', started_at=old_dt, company_id=company.id, created_at=old_dt, updated_at=old_dt))
db.session.add(NotificationLog(invoice_id=invoice.id, channel='mail', status='done', created_at=old_dt, updated_at=old_dt))
db.session.add(MailDelivery(invoice_id=invoice.id, recipient='a@example.com', status='sent', created_at=old_dt, updated_at=old_dt))
db.session.add(SyncEvent(invoice_id=invoice.id, status='ok', created_at=old_dt, updated_at=old_dt))
db.session.commit()
response = auth_client.post('/admin/logs/cleanup', data={'days': 90}, follow_redirects=True)
assert response.status_code == 200
body = response.get_data(as_text=True)
assert 'Usunięto stare logi starsze niż 90 dni' in body
with app.app_context():
assert AuditLog.query.filter_by(action='old').count() == 0
assert SyncLog.query.filter_by(sync_type='manual').count() == 0
assert NotificationLog.query.count() == 0
assert MailDelivery.query.count() == 0
def test_database_backup_download(auth_client, app):
response = auth_client.post('/admin/database/backup')
assert response.status_code == 200
assert 'attachment;' in response.headers.get('Content-Disposition', '')
assert 'db_backup_' in response.headers.get('Content-Disposition', '')
with app.app_context():
backup_dir = Path(app.config['BACKUP_PATH'])
assert any(path.name.startswith('db_backup_') for path in backup_dir.iterdir())
def test_clear_mock_data_removes_legacy_mock_invoices(app, auth_client):
from app.models.invoice import Invoice, InvoiceStatus, InvoiceType
with app.app_context():
company = Company.query.first()
company_id = company.id
rows = [
Invoice(company_id=company_id, ksef_number='KSEF/MOCK/C1/2026/10001', invoice_number='FV/1/001/2026', contractor_name='Firma 1-1', issue_date=datetime.utcnow().date(), net_amount=1, vat_amount=0.23, gross_amount=1.23, invoice_type=InvoiceType.SALE, status=InvoiceStatus.NEW, source='ksef', issued_status='received', external_metadata={'source': 'mock', 'sequence': 1, 'company_id': company_id}),
Invoice(company_id=company_id, ksef_number='KSEF/MOCK/C1/2026/10002', invoice_number='FV/1/002/2026', contractor_name='Firma 1-2', issue_date=datetime.utcnow().date(), net_amount=1, vat_amount=0.23, gross_amount=1.23, invoice_type=InvoiceType.SALE, status=InvoiceStatus.NEW, source='ksef', issued_status='received', external_metadata={'source': 'mock', 'sequence': 2, 'company_id': company_id}),
Invoice(company_id=company_id, ksef_number='KSEF/MOCK/C1/2026/10003', invoice_number='FV/1/003/2026', contractor_name='Firma 1-3', issue_date=datetime.utcnow().date(), net_amount=1, vat_amount=0.23, gross_amount=1.23, invoice_type=InvoiceType.SALE, status=InvoiceStatus.NEW, source='ksef', issued_status='received', external_metadata={'source': 'mock', 'sequence': 3, 'company_id': company_id}),
Invoice(company_id=company_id, ksef_number='KSEF/MOCK/C1/2026/10004', invoice_number='FV/1/004/2026', contractor_name='Firma 1-4', issue_date=datetime.utcnow().date(), net_amount=1, vat_amount=0.23, gross_amount=1.23, invoice_type=InvoiceType.SALE, status=InvoiceStatus.NEW, source='ksef', issued_status='received', external_metadata={'source': 'mock', 'sequence': 4, 'company_id': company_id}),
Invoice(company_id=company_id, ksef_number='KSEF/MOCK/C1/2026/10005', invoice_number='FV/1/005/2026', contractor_name='Firma 1-5', issue_date=datetime.utcnow().date(), net_amount=1, vat_amount=0.23, gross_amount=1.23, invoice_type=InvoiceType.SALE, status=InvoiceStatus.NEW, source='ksef', issued_status='received', external_metadata={'source': 'mock', 'sequence': 5, 'company_id': company_id}),
Invoice(company_id=company_id, ksef_number='PENDING/FV/2026/03/0002', invoice_number='FV/2026/03/0002', contractor_name='aaaa', issue_date=datetime.utcnow().date(), net_amount=1, vat_amount=0.23, gross_amount=1.23, invoice_type=InvoiceType.SALE, status=InvoiceStatus.NEW, source='issued', issued_status='draft', external_metadata={}),
Invoice(company_id=company_id, ksef_number='NFZ-PENDING/FV/2026/03/0003', invoice_number='FV/2026/03/0003', contractor_name='NFZ', issue_date=datetime.utcnow().date(), net_amount=1, vat_amount=0.23, gross_amount=1.23, invoice_type=InvoiceType.SALE, status=InvoiceStatus.NEW, source='nfz', issued_status='draft', external_metadata={'nfz': {'x': 1}}),
Invoice(company_id=company_id, ksef_number='KSEF/MOCK/ISSUED/C1/2026/1773052148', invoice_number='FV/2026/03/0004', contractor_name='aaaa', issue_date=datetime.utcnow().date(), net_amount=1, vat_amount=0.23, gross_amount=1.23, invoice_type=InvoiceType.SALE, status=InvoiceStatus.NEW, source='issued', issued_status='issued_mock', external_metadata={}),
]
db.session.add_all(rows)
AppSetting.set(f'company.{company_id}.ksef.mock_mode', 'true')
db.session.commit()
assert Invoice.query.count() == 8
assert Invoice.query.filter(Invoice.ksef_number.like('%MOCK%')).count() == 6
response = auth_client.post('/admin/mock-data/clear', follow_redirects=True)
assert response.status_code == 200
body = response.get_data(as_text=True)
assert 'Usunięto dane mock: faktury 6' in body
with app.app_context():
remaining_numbers = [row[0] for row in db.session.query(Invoice.invoice_number).order_by(Invoice.id).all()]
assert remaining_numbers == ['FV/2026/03/0002', 'FV/2026/03/0003']
assert AppSetting.get(f'company.{company_id}.ksef.mock_mode') == 'false'
def test_save_ceidg_settings_persists_api_key(app, auth_client):
long_api_key = 'A1B2C3D4' * 32
response = auth_client.post(
'/admin/ceidg/save',
data={'environment': 'test', 'api_key': long_api_key},
follow_redirects=True,
)
assert response.status_code == 200
with app.app_context():
assert AppSetting.get('ceidg.environment') == 'test'
assert AppSetting.get('ceidg.api_key', decrypt=True) == long_api_key
stored = AppSetting.query.filter_by(key='ceidg.api_key').first()
assert stored is not None
assert stored.is_encrypted is True
assert stored.value != long_api_key

17
tests/test_api.py Normal file
View File

@@ -0,0 +1,17 @@
from app.models.company import Company
from app.services.sync_service import SyncService
def test_health(client):
response = client.get('/api/health')
assert response.status_code == 200
assert response.json['db'] == 'ok'
def test_api_invoices(auth_client, app):
with app.app_context():
company = Company.query.first()
SyncService(company).run_manual_sync()
response = auth_client.get('/api/invoices')
assert response.status_code == 200
assert 'items' in response.json

10
tests/test_auth.py Normal file
View File

@@ -0,0 +1,10 @@
def test_login_page(client):
response = client.get('/auth/login')
assert response.status_code == 200
assert b'Logowanie' in response.data
def test_login_success(client):
response = client.post('/auth/login', data={'email': 'admin@example.com', 'password': 'Admin123!'}, follow_redirects=True)
assert response.status_code == 200
assert b'Dashboard' in response.data

68
tests/test_ceidg_service.py Normal file
View File

@@ -0,0 +1,68 @@
import json
from app.models.setting import AppSetting
from app.services.ceidg_service import CeidgService
def test_ceidg_service_parses_warehouse_payload(app):
payload = {
'firmy': [
{
'id': '9D2531B1-6DED-4538-95EA-22FF2C7D2E20',
'nazwa': 'Adam IntegracjaMGMF',
'adresDzialalnosci': {
'ulica': 'ul. Zwierzyniecka',
'budynek': '1',
'miasto': 'Białystok',
'kod': '15-333',
},
'wlasciciel': {
'imie': 'Adam',
'nazwa': 'IntegracjaMGMF',
'nip': '3563457932',
'regon': '518155359',
},
}
]
}
with app.app_context():
parsed = CeidgService()._parse_payload(json.dumps(payload), '3563457932')
assert parsed == {
'name': 'Adam IntegracjaMGMF',
'regon': '518155359',
'address': 'ul. Zwierzyniecka 1, 15-333 Białystok',
'tax_id': '3563457932',
}
def test_ceidg_service_uses_bearer_authorization(app):
with app.app_context():
AppSetting.set('ceidg.api_key', 'jwt-token', encrypt=True)
headers = CeidgService._headers()
assert headers['Authorization'] == 'Bearer jwt-token'
def test_admin_company_fetch_from_ceidg_requires_only_nip(app, auth_client, monkeypatch):
lookup = {'ok': True, 'name': 'Test CEIDG', 'tax_id': '1234567890', 'regon': '123456789', 'address': 'Warszawa'}
monkeypatch.setattr('app.admin.routes.CeidgService.fetch_company', lambda self, identifier=None, **kwargs: lookup)
response = auth_client.post(
'/admin/companies/new',
data={
'name': '',
'tax_id': '1234567890',
'regon': '',
'address': '',
'sync_interval_minutes': '60',
'fetch_submit': '1',
},
follow_redirects=True,
)
body = response.get_data(as_text=True)
assert response.status_code == 200
assert 'Pobrano dane firmy z CEIDG.' in body
assert 'Test CEIDG' in body

45
tests/test_dashboard.py Normal file
View File

@@ -0,0 +1,45 @@
from datetime import date, datetime, timedelta
from app.extensions import db
from app.models.company import Company
from app.models.invoice import Invoice, InvoiceStatus, InvoiceType
def test_dashboard_recent_invoices_uses_10_items_per_page(auth_client, app):
with app.app_context():
company = Company.query.first()
now = datetime.utcnow()
for index in range(12):
invoice = Invoice(
company_id=company.id,
ksef_number=f'KSEF-DASH-{index}',
invoice_number=f'FV/DASH/{index}',
contractor_name=f'Kontrahent {index}',
issue_date=date(2026, 3, 12) - timedelta(days=index),
received_date=date(2026, 3, 12) - timedelta(days=index),
net_amount=100 + index,
vat_amount=23,
gross_amount=123 + index,
invoice_type=InvoiceType.PURCHASE,
status=InvoiceStatus.NEW,
source='ksef',
created_at=now - timedelta(minutes=index),
)
db.session.add(invoice)
db.session.commit()
response_page_1 = auth_client.get('/')
assert response_page_1.status_code == 200
assert response_page_1.data.count(b'btn btn-outline-primary">Szczeg') == 10
assert b'FV/DASH/0' in response_page_1.data
assert b'FV/DASH/9' in response_page_1.data
assert b'FV/DASH/10' not in response_page_1.data
assert b'FV/DASH/11' not in response_page_1.data
assert b'dashboard_page=2' in response_page_1.data
response_page_2 = auth_client.get('/?dashboard_page=2')
assert response_page_2.status_code == 200
assert response_page_2.data.count(b'btn btn-outline-primary">Szczeg') == 2
assert b'FV/DASH/10' in response_page_2.data
assert b'FV/DASH/11' in response_page_2.data
assert b'FV/DASH/9' not in response_page_2.data

160
tests/test_invoices.py Normal file
View File

@@ -0,0 +1,160 @@
from app.models.company import Company
from app.models.invoice import Invoice
from app.services.sync_service import SyncService
def test_invoice_list(auth_client, app):
with app.app_context():
company = Company.query.first()
SyncService(company).run_manual_sync()
response = auth_client.get('/invoices/')
assert response.status_code == 200
assert b'Faktury' in response.data
def test_invoice_pdf(auth_client, app):
with app.app_context():
company = Company.query.first()
SyncService(company).run_manual_sync()
invoice = Invoice.query.first()
response = auth_client.get(f'/invoices/{invoice.id}/pdf')
assert response.status_code == 200
assert response.mimetype == 'application/pdf'
def test_issued_form_uses_quick_add_modals(auth_client):
response = auth_client.get('/invoices/issued/new')
assert response.status_code == 200
assert b'customerQuickAddModal' in response.data
assert b'productQuickAddModal' in response.data
assert b'Szybkie dodanie klienta' not in response.data
def test_invoice_list_shows_only_incoming(auth_client, app):
from datetime import date
from app.extensions import db
from app.models.company import Company
from app.models.invoice import Invoice, InvoiceStatus, InvoiceType
with app.app_context():
company = Company.query.first()
sale = Invoice(
company_id=company.id,
ksef_number='SALE-HIDDEN',
invoice_number='FV/SALE/1',
contractor_name='Sprzedaz Sp z o.o.',
issue_date=date(2026, 3, 1),
received_date=date(2026, 3, 1),
net_amount=100,
vat_amount=23,
gross_amount=123,
invoice_type=InvoiceType.SALE,
status=InvoiceStatus.NEW,
source='issued',
issued_status='draft',
)
purchase = Invoice(
company_id=company.id,
ksef_number='PURCHASE-VISIBLE',
invoice_number='FV/PUR/1',
contractor_name='Dostawca Sp z o.o.',
issue_date=date(2026, 3, 1),
received_date=date(2026, 3, 1),
net_amount=200,
vat_amount=46,
gross_amount=246,
invoice_type=InvoiceType.PURCHASE,
status=InvoiceStatus.NEW,
source='ksef',
)
db.session.add_all([sale, purchase])
db.session.commit()
response = auth_client.get('/invoices/')
assert response.status_code == 200
assert b'FV/PUR/1' in response.data
assert b'FV/SALE/1' not in response.data
assert b'Faktury otrzymane' in response.data
def test_resolve_payment_details_reads_metadata_and_xml(app, tmp_path):
from datetime import date
from app.extensions import db
from app.models.company import Company
from app.models.invoice import Invoice, InvoiceStatus, InvoiceType
from app.services.invoice_service import InvoiceService
xml_path = tmp_path / 'invoice.xml'
xml_path.write_text('''<?xml version="1.0" encoding="UTF-8"?>
<Faktura xmlns="http://example.com/fa">
<Platnosc>
<FormaPlatnosci>6</FormaPlatnosci>
<RachunekBankowy>
<NrRB>12 3456 7890 1234 5678 9012 3456</NrRB>
<NazwaBanku>Bank Testowy</NazwaBanku>
</RachunekBankowy>
<TerminPlatnosci>
<Termin>2026-03-31</Termin>
</TerminPlatnosci>
</Platnosc>
</Faktura>''', encoding='utf-8')
with app.app_context():
company = Company.query.first()
invoice = Invoice(
company_id=company.id,
ksef_number='KSEF/TEST/1',
invoice_number='FV/TEST/1',
contractor_name='Dostawca',
issue_date=date(2026, 3, 12),
received_date=date(2026, 3, 12),
net_amount=100,
vat_amount=23,
gross_amount=123,
invoice_type=InvoiceType.PURCHASE,
status=InvoiceStatus.NEW,
source='ksef',
xml_path=str(xml_path),
external_metadata={'payment_form_label': 'przelew'},
)
db.session.add(invoice)
db.session.commit()
details = InvoiceService().resolve_payment_details(invoice)
assert details['payment_form_code'] == '6'
assert details['payment_form_label'] == 'przelew'
assert details['bank_account'] == '12345678901234567890123456'
assert details['bank_name'] == 'Bank Testowy'
assert details['payment_due_date'] == '2026-03-31'
def test_purchase_invoice_does_not_fallback_to_company_bank_account(app):
from datetime import date
from app.extensions import db
from app.models.company import Company
from app.models.invoice import Invoice, InvoiceStatus, InvoiceType
from app.services.invoice_service import InvoiceService
with app.app_context():
company = Company.query.first()
company.bank_account = '11 1111 1111 1111 1111 1111 1111'
invoice = Invoice(
company_id=company.id,
ksef_number='KSEF/TEST/2',
invoice_number='FV/TEST/2',
contractor_name='Dostawca',
issue_date=date(2026, 3, 12),
received_date=date(2026, 3, 12),
net_amount=100,
vat_amount=23,
gross_amount=123,
invoice_type=InvoiceType.PURCHASE,
status=InvoiceStatus.NEW,
source='ksef',
)
db.session.add(invoice)
db.session.commit()
assert InvoiceService()._resolve_seller_bank_account(invoice) == ''

125
tests/test_ksef_xml_generator.py Normal file
View File

@@ -0,0 +1,125 @@
from datetime import date
import xml.etree.ElementTree as ET
from app.models.catalog import InvoiceLine
from app.models.company import Company
from app.models.invoice import Invoice, InvoiceType
from app.services.invoice_service import InvoiceService
NS = {'fa': 'http://crd.gov.pl/wzor/2025/06/25/13775/'}
def test_render_structured_xml_uses_fa3_schema_and_split_payment(app):
with app.app_context():
company = Company.query.first()
company.name = 'Test Co Sp. z o.o.'
company.tax_id = '5250000001'
company.address = 'ul. Testowa 1, 00-001 Warszawa'
company.bank_account = '11 1111 1111 1111 1111 1111 1111'
invoice = Invoice(
company_id=company.id,
company=company,
ksef_number='DRAFT/1',
invoice_number='FV/1/2026',
contractor_name='Klient Sp. z o.o.',
contractor_nip='5260000002',
contractor_address='ul. Odbiorcy 2, 00-002 Warszawa',
issue_date=date(2026, 3, 12),
net_amount=100,
vat_amount=23,
gross_amount=123,
invoice_type=InvoiceType.SALE,
split_payment=True,
currency='PLN',
seller_bank_account='11 1111 1111 1111 1111 1111 1111',
source='issued',
external_metadata={},
)
lines = [
InvoiceLine(
description='Usługa abonamentowa',
quantity=1,
unit='usł.',
unit_net=100,
vat_rate=23,
net_amount=100,
vat_amount=23,
gross_amount=123,
)
]
xml = InvoiceService().render_structured_xml(invoice, lines=lines, nfz_meta={})
root = ET.fromstring(xml)
assert root.tag == '{http://crd.gov.pl/wzor/2025/06/25/13775/}Faktura'
assert root.findtext('fa:Naglowek/fa:KodFormularza', namespaces=NS) == 'FA'
kod = root.find('fa:Naglowek/fa:KodFormularza', NS)
assert kod is not None
assert kod.attrib['kodSystemowy'] == 'FA (3)'
assert kod.attrib['wersjaSchemy'] == '1-0E'
assert root.findtext('fa:Naglowek/fa:WariantFormularza', namespaces=NS) == '3'
assert root.findtext('fa:Fa/fa:P_2', namespaces=NS) == 'FV/1/2026'
assert root.findtext('fa:Fa/fa:Adnotacje/fa:P_18A', namespaces=NS) == '1'
assert root.findtext('fa:Fa/fa:P_13_1', namespaces=NS) == '100.00'
assert root.findtext('fa:Fa/fa:P_14_1', namespaces=NS) == '23.00'
assert root.findtext('fa:Fa/fa:P_15', namespaces=NS) == '123.00'
assert root.findtext('fa:Fa/fa:Platnosc/fa:FormaPlatnosci', namespaces=NS) == '6'
assert root.findtext('fa:Fa/fa:Platnosc/fa:RachunekBankowy/fa:NrRB', namespaces=NS) == '11111111111111111111111111'
assert root.findtext('fa:Fa/fa:FaWiersz/fa:P_8A', namespaces=NS) == 'usł.'
assert root.findtext('fa:Fa/fa:FaWiersz/fa:P_8B', namespaces=NS) == '1'
def test_render_structured_xml_maps_nfz_metadata_to_dodatkowy_opis(app):
with app.app_context():
company = Company.query.first()
invoice = Invoice(
company_id=company.id,
company=company,
ksef_number='DRAFT/2',
invoice_number='FV/NFZ/1/2026',
contractor_name='NFZ',
contractor_nip='1234567890',
issue_date=date(2026, 3, 12),
net_amount=200,
vat_amount=46,
gross_amount=246,
invoice_type=InvoiceType.SALE,
split_payment=False,
currency='PLN',
source='nfz',
external_metadata={},
)
lines = [
InvoiceLine(
description='Świadczenie medyczne',
quantity=2,
unit='usł.',
unit_net=100,
vat_rate=23,
net_amount=200,
vat_amount=46,
gross_amount=246,
)
]
nfz_meta = {
'recipient_branch_id': '01',
'settlement_from': '2026-03-01',
'settlement_to': '2026-03-31',
'provider_identifier': 'NFZ-ABC',
'service_code': 'SVC-1',
'contract_number': 'UM-2026-01',
'template_identifier': 'TPL-77',
}
xml = InvoiceService().render_structured_xml(invoice, lines=lines, nfz_meta=nfz_meta)
root = ET.fromstring(xml)
dodatki = root.findall('fa:Fa/fa:DodatkowyOpis', NS)
pairs = {item.findtext('fa:Klucz', namespaces=NS): item.findtext('fa:Wartosc', namespaces=NS) for item in dodatki}
assert pairs['IDWew'] == '01'
assert pairs['P_6_Od'] == '2026-03-01'
assert pairs['P_6_Do'] == '2026-03-31'
assert pairs['NrUmowy'] == 'UM-2026-01'
assert root.findtext('fa:Podmiot3/fa:DaneIdentyfikacyjne/fa:IDWew', namespaces=NS) == '01'

35
tests/test_nfz.py Normal file
View File

@@ -0,0 +1,35 @@
from decimal import Decimal
from app.extensions import db
from app.models.catalog import Customer, Product
from app.models.company import Company
from app.models.setting import AppSetting
def test_nfz_index_loads_with_module_enabled(auth_client, app):
with app.app_context():
company = Company.query.first()
AppSetting.set(f'company.{company.id}.modules.nfz_enabled', 'true')
db.session.add(Customer(company_id=company.id, name='NFZ Client', tax_id='1070001057', is_active=True))
db.session.add(Product(company_id=company.id, name='Swiadczenie', unit='usl', net_price=Decimal('100.00'), vat_rate=Decimal('8.00'), is_active=True))
db.session.commit()
response = auth_client.get('/nfz/')
assert response.status_code == 200
assert b'NFZ' in response.data
def test_nfz_index_uses_shared_quick_add_modals(auth_client, app):
with app.app_context():
company = Company.query.first()
AppSetting.set(f'company.{company.id}.modules.nfz_enabled', 'true')
if not Customer.query.filter_by(company_id=company.id, name='NFZ Modal Client').first():
db.session.add(Customer(company_id=company.id, name='NFZ Modal Client', tax_id='1070001057', is_active=True))
if not Product.query.filter_by(company_id=company.id, name='NFZ Modal Service').first():
db.session.add(Product(company_id=company.id, name='NFZ Modal Service', unit='usl', net_price=Decimal('50.00'), vat_rate=Decimal('8.00'), is_active=True))
db.session.commit()
response = auth_client.get('/nfz/')
assert response.status_code == 200
assert b'customerQuickAddModal' in response.data
assert b'productQuickAddModal' in response.data
assert b'Szybkie dodanie klienta' not in response.data

99
tests/test_nfz_production.py Normal file
View File

@@ -0,0 +1,99 @@
from decimal import Decimal
from app.extensions import db
from app.models.catalog import Customer, Product, InvoiceLine
from app.models.company import Company
from app.models.invoice import Invoice, InvoiceStatus, InvoiceType
from app.models.setting import AppSetting
from app.services.invoice_service import InvoiceService
def _enable_nfz(app):
with app.app_context():
company = Company.query.first()
AppSetting.set(f'company.{company.id}.modules.nfz_enabled', 'true')
customer = Customer.query.filter_by(company_id=company.id, tax_id='1070001057').first()
if not customer:
customer = Customer(company_id=company.id, name='NFZ Client', tax_id='1070001057', is_active=True)
db.session.add(customer)
product = Product.query.filter_by(company_id=company.id, name='Swiadczenie NFZ').first()
if not product:
product = Product(company_id=company.id, name='Swiadczenie NFZ', unit='usl', net_price=Decimal('100.00'), vat_rate=Decimal('8.00'), is_active=True)
db.session.add(product)
db.session.commit()
return company.id, customer.id, product.id
def test_duplicate_nfz_redirects_to_nfz_form(auth_client, app):
company_id, customer_id, product_id = _enable_nfz(app)
with app.app_context():
invoice = Invoice(
company_id=company_id,
customer_id=customer_id,
ksef_number='NFZ-PENDING/1',
invoice_number='FV/NFZ/1',
contractor_name='Narodowy Fundusz Zdrowia - Slaski OW NFZ',
contractor_nip='1070001057',
issue_date=InvoiceService.today_date(),
received_date=InvoiceService.today_date(),
net_amount=Decimal('100.00'),
vat_amount=Decimal('8.00'),
gross_amount=Decimal('108.00'),
invoice_type=InvoiceType.SALE,
status=InvoiceStatus.NEW,
source='nfz',
issued_status='draft',
external_metadata={'nfz': {'recipient_branch_id': '1070001057-00122', 'settlement_from': '2026-02-01', 'settlement_to': '2026-02-28', 'template_identifier': 'TMP', 'provider_identifier': 'SWD1', 'service_code': '01.02', 'contract_number': 'C/1', 'nfz_schema': 'FA(3)'}},
)
db.session.add(invoice)
db.session.flush()
db.session.add(InvoiceLine(invoice_id=invoice.id, product_id=product_id, description='Swiadczenie NFZ', quantity=1, unit='usl', unit_net=Decimal('100.00'), vat_rate=Decimal('8.00'), net_amount=Decimal('100.00'), vat_amount=Decimal('8.00'), gross_amount=Decimal('108.00')))
db.session.commit()
invoice_id = invoice.id
response = auth_client.get(f'/invoices/{invoice_id}/duplicate', follow_redirects=False)
assert response.status_code == 302
assert f'/nfz/?duplicate_id={invoice_id}' in response.headers['Location']
form_response = auth_client.get(response.headers['Location'])
body = form_response.get_data(as_text=True)
assert form_response.status_code == 200
assert 'FV/NFZ/1/COPY' in body
assert 'SWD1' in body
assert 'C/1' in body
def test_build_ksef_payload_contains_nfz_xml(app):
company_id, customer_id, product_id = _enable_nfz(app)
with app.app_context():
company = Company.query.first()
invoice = Invoice(
company_id=company_id,
company=company,
customer_id=customer_id,
ksef_number='NFZ-PENDING/2',
invoice_number='FV/NFZ/2',
contractor_name='Narodowy Fundusz Zdrowia - Slaski OW NFZ',
contractor_nip='1070001057',
issue_date=InvoiceService.today_date(),
received_date=InvoiceService.today_date(),
net_amount=Decimal('100.00'),
vat_amount=Decimal('8.00'),
gross_amount=Decimal('108.00'),
invoice_type=InvoiceType.SALE,
status=InvoiceStatus.NEW,
source='nfz',
issued_status='draft',
external_metadata={'nfz': {'recipient_branch_id': '1070001057-00122', 'settlement_from': '2026-02-01', 'settlement_to': '2026-02-28', 'template_identifier': 'TMP', 'provider_identifier': 'SWD1', 'service_code': '01.02', 'contract_number': 'C/1', 'nfz_schema': 'FA(3)'}},
)
db.session.add(invoice)
db.session.flush()
db.session.add(InvoiceLine(invoice_id=invoice.id, product_id=product_id, description='Swiadczenie NFZ', quantity=1, unit='usl', unit_net=Decimal('100.00'), vat_rate=Decimal('8.00'), net_amount=Decimal('100.00'), vat_amount=Decimal('8.00'), gross_amount=Decimal('108.00')))
db.session.commit()
payload = InvoiceService().build_ksef_payload(invoice)
assert payload['schemaVersion'] == 'FA(3)'
assert '<IDWew>1070001057-00122</IDWew>' in payload['xml_content']
assert '<P_6_Od>2026-02-01</P_6_Od>' in payload['xml_content']
assert '<P_6_Do>2026-02-28</P_6_Do>' in payload['xml_content']
assert 'NrUmowy' in payload['xml_content']

108
tests/test_settings_ksef.py Normal file
View File

@@ -0,0 +1,108 @@
import base64
from io import BytesIO
from app.models.company import Company
from app.models.setting import AppSetting
from app.services.ksef_service import RequestsKSeFAdapter
def test_save_ksef_token_and_certificate(auth_client, app):
response = auth_client.post(
'/settings/',
data={
'ksef-base_url': ' https://api.ksef.mf.gov.pl ',
'ksef-auth_mode': 'token',
'ksef-client_id': ' client-1 ',
'ksef-token': ' secret-token ',
'ksef-submit': '1',
},
content_type='multipart/form-data',
follow_redirects=True,
)
assert response.status_code == 200
certificate_bytes = b"""-----BEGIN CERTIFICATE-----
TEST
-----END CERTIFICATE-----
"""
response = auth_client.post(
'/settings/',
data={
'ksef-base_url': 'https://api.ksef.mf.gov.pl/docs/v2',
'ksef-auth_mode': 'certificate',
'ksef-client_id': 'client-2',
'ksef-token': '',
'ksef-certificate_file': (BytesIO(certificate_bytes), 'cert.pem'),
'ksef-submit': '1',
},
content_type='multipart/form-data',
follow_redirects=True,
)
assert response.status_code == 200
with app.app_context():
company = Company.query.first()
prefix = f'company.{company.id}.ksef'
assert AppSetting.get(f'{prefix}.base_url') == 'https://api.ksef.mf.gov.pl/docs/v2'
assert AppSetting.get(f'{prefix}.auth_mode') == 'certificate'
assert AppSetting.get(f'{prefix}.client_id') == 'client-2'
assert AppSetting.get(f'{prefix}.token', decrypt=True) == 'secret-token'
assert AppSetting.get(f'{prefix}.certificate_name') == 'cert.pem'
stored = AppSetting.get(f'{prefix}.certificate_data', decrypt=True)
assert base64.b64decode(stored) == certificate_bytes
def test_ksef_settings_page_shows_saved_status(auth_client, app):
with app.app_context():
company = Company.query.first()
AppSetting.set(f'company.{company.id}.ksef.token', 'abc', encrypt=True)
AppSetting.set(f'company.{company.id}.ksef.certificate_name', 'cert.pem')
AppSetting.set(f'company.{company.id}.ksef.certificate_data', 'Y2VydA==', encrypt=True)
AppSetting.set(f'company.{company.id}.ksef.mock_mode', 'true')
from app.extensions import db
db.session.commit()
response = auth_client.get('/settings/')
body = response.get_data(as_text=True)
assert response.status_code == 200
assert 'Token KSeF jest zapisany w konfiguracji tej firmy.' in body
assert 'Certyfikat KSeF jest zapisany w konfiguracji tej firmy.' in body
assert 'Wgrany plik: cert.pem' in body
def test_requests_adapter_uses_supported_invoice_endpoints(app, monkeypatch):
calls = []
def fake_request(self, method, path, params=None, json=None, accept='application/json'):
calls.append((method, path, params, json, accept))
if path == '/invoices/query/metadata':
return {
'invoices': [
{
'ksefNumber': '5555555555-20250828-010080615740-E4',
'invoiceNumber': 'FV/1',
'issueDate': '2025-08-27',
'acquisitionDate': '2025-08-28T09:22:56+00:00',
'seller': {'nip': '5555555555', 'name': 'Test Company 1'},
'netAmount': 100,
'vatAmount': 23,
'grossAmount': 123,
}
]
}
if path == '/invoices/ksef/5555555555-20250828-010080615740-E4':
return '<Invoice/>'
raise AssertionError(path)
with app.app_context():
company = Company.query.first()
AppSetting.set(f'company.{company.id}.ksef.mock_mode', 'false')
from app.extensions import db
db.session.commit()
monkeypatch.setattr(RequestsKSeFAdapter, '_request', fake_request)
documents = RequestsKSeFAdapter(company_id=company.id).list_documents()
assert len(documents) == 1
assert calls[0][1] == '/invoices/query/metadata'
assert calls[1][1] == '/invoices/ksef/5555555555-20250828-010080615740-E4'
assert calls[1][4] == 'application/xml'

11
tests/test_sync.py Normal file
View File

@@ -0,0 +1,11 @@
from app.models.company import Company
from app.models.invoice import Invoice
from app.services.sync_service import SyncService
def test_sync_creates_invoices(app):
with app.app_context():
company = Company.query.first()
log = SyncService(company).run_manual_sync()
assert log.status == 'finished'
assert Invoice.query.filter_by(company_id=company.id).count() > 0