Mateusz Gruszczyński
319 lines
13 KiB
Python
319 lines
13 KiB
Python
from app import create_app
|
||
from app.database import initialize_database, migrate_database
|
||
from app.services import calculate_costs, invoice_period
|
||
from datetime import datetime
|
||
|
||
|
||
def make_app(tmp_path, name):
|
||
return create_app({"TESTING": True, "SQLALCHEMY_DATABASE_URI": f"sqlite:///{tmp_path / name}"})
|
||
|
||
def test_health(tmp_path):
|
||
app=make_app(tmp_path, "test.db")
|
||
assert app.test_client().get('/health').json=={"status":"ok"}
|
||
|
||
def test_costs_50_percent_vat():
|
||
result=calculate_costs(123,23,50)
|
||
assert float(result['net'])==100.0
|
||
assert float(result['deductible_vat'])==11.5
|
||
assert float(result['final_cost'])==111.5
|
||
|
||
def test_invoice_period():
|
||
assert invoice_period(datetime(2026,7,15),15).endswith('01–15')
|
||
assert invoice_period(datetime(2026,7,16),15).endswith('16–koniec')
|
||
|
||
def test_admin_settings_ajax_has_working_route(tmp_path):
|
||
app=make_app(tmp_path, "admin.db")
|
||
client=app.test_client()
|
||
client.post('/login',data={'email':'admin@example.com','password':'admin123!'})
|
||
response=client.post('/admin',data={'action':'settings','name':'Firma','entity_type':'JDG','vat_rate':'23','vat_deduction_percent':'50','invoice_split_day':'15','invoice_split_enabled':'on','region':'mazowieckie','theme':'darkly'},headers={'X-Requested-With':'XMLHttpRequest'})
|
||
assert response.status_code==200
|
||
assert response.json['ok'] is True
|
||
|
||
def test_ure_brand_aggregation(tmp_path):
|
||
from app.integrations.ure import aggregate_ure_companies
|
||
app=make_app(tmp_path, "ure.db")
|
||
with app.app_context():
|
||
rows=[{"nazwa":"ORLEN S.A.","nip":"7740001454","regon":"610188201","nazwaStacji":"100","wojewodztwo":"mazowieckie","benzynySilnikowe":True},
|
||
{"nazwa":"ORLEN S.A.","nip":"7740001454","regon":"610188201","nazwaStacji":"101","wojewodztwo":"śląskie","olejeNapedowe":True}]
|
||
result=aggregate_ure_companies(rows)
|
||
assert len(result)==1
|
||
assert result[0]["brand_name"]=="ORLEN"
|
||
assert result[0]["station_count"]==2
|
||
|
||
def test_ure_duplicate_dkn_is_merged(tmp_path):
|
||
from app.integrations.ure import aggregate_ure_companies
|
||
app=make_app(tmp_path, "dup.db")
|
||
rows=[
|
||
{"dkn":977,"nazwa":"Test Sp. z o.o.","nip":"123","nazwaStacji":"Punkt paliw","infraUlica":"A","infraNrLokalu":"1","infraPoczta":"Miasto","benzynySilnikowe":True},
|
||
{"dkn":977,"nazwa":"Test Sp. z o.o.","nip":"123","nazwaStacji":"Punkt paliw i LPG","infraUlica":"A","infraNrLokalu":"1","infraPoczta":"Miasto","gazPlynnyLPG":True},
|
||
]
|
||
with app.app_context():
|
||
result=aggregate_ure_companies(rows)
|
||
assert len(result)==1
|
||
assert result[0]["station_count"]==1
|
||
assert len(result[0]["points"])==1
|
||
assert result[0]["points"][0]["has_petrol"] is True
|
||
assert result[0]["points"][0]["has_lpg"] is True
|
||
|
||
def test_company_and_card_forms_use_post(tmp_path):
|
||
app=make_app(tmp_path, "forms.db")
|
||
client=app.test_client()
|
||
client.post('/login',data={'email':'admin@example.com','password':'admin123!'})
|
||
html=client.get('/admin/companies').get_data(as_text=True)
|
||
assert 'action="/api/companies" method="post"' in html
|
||
assert 'action="/api/fuel-cards" method="post"' in html
|
||
assert 'Nazwa firmy' in html
|
||
assert 'Operator karty' in html
|
||
|
||
|
||
def test_cannot_delete_last_admin(tmp_path):
|
||
app=make_app(tmp_path, "delete.db")
|
||
client=app.test_client()
|
||
client.post('/api/auth/login',data={'email':'admin@example.com','password':'admin123!'})
|
||
with app.app_context():
|
||
from app.models import User
|
||
admin=User.query.filter_by(email='admin@example.com').first()
|
||
admin_id=admin.id
|
||
response=client.delete(f'/api/users/{admin_id}')
|
||
assert response.status_code == 400
|
||
assert response.json['ok'] is False
|
||
|
||
|
||
def test_cli_reset_admin_password(tmp_path):
|
||
app=make_app(tmp_path, "cli.db")
|
||
runner=app.test_cli_runner()
|
||
result=runner.invoke(args=['reset-admin-password','--email','admin@example.com','--password','NoweHaslo123!'])
|
||
assert result.exit_code == 0
|
||
client=app.test_client()
|
||
response=client.post('/api/auth/login',data={'email':'admin@example.com','password':'NoweHaslo123!'})
|
||
assert response.status_code == 200
|
||
|
||
|
||
def test_html_cache_headers_and_static_versioning(tmp_path):
|
||
app=make_app(tmp_path, "cache.db")
|
||
client=app.test_client()
|
||
response=client.get('/login')
|
||
assert response.headers['Cache-Control'] == 'no-store, private, must-revalidate'
|
||
assert 'Pragma' not in response.headers
|
||
assert 'Expires' not in response.headers
|
||
html=response.get_data(as_text=True)
|
||
assert '/static/css/layout.css?v=' in html
|
||
assert '/static/js/core.js?v=' in html
|
||
|
||
|
||
def test_css_cache_for_30_days(tmp_path):
|
||
app=make_app(tmp_path, "static.db")
|
||
response=app.test_client().get('/static/css/layout.css')
|
||
assert response.status_code == 200
|
||
assert response.headers['Cache-Control'] == 'public, max-age=2592000, immutable'
|
||
|
||
|
||
def test_create_app_initializes_database(tmp_path):
|
||
from sqlalchemy import inspect
|
||
from app.extensions import db
|
||
|
||
app = create_app({"TESTING": True, "SQLALCHEMY_DATABASE_URI": f"sqlite:///{tmp_path / 'empty.db'}"})
|
||
with app.app_context():
|
||
tables = inspect(db.engine).get_table_names()
|
||
assert "user" in tables
|
||
assert "schema_migration" in tables
|
||
|
||
|
||
def test_migration_runner_creates_registry_without_legacy_migrations(tmp_path):
|
||
from sqlalchemy import inspect
|
||
from app.extensions import db
|
||
|
||
app = make_app(tmp_path, "migrations.db")
|
||
with app.app_context():
|
||
assert migrate_database() == []
|
||
assert "schema_migration" in inspect(db.engine).get_table_names()
|
||
assert migrate_database() == []
|
||
|
||
def test_global_vat_override_changes_50_percent_deduction(tmp_path):
|
||
from decimal import Decimal
|
||
from datetime import date
|
||
from app.extensions import db
|
||
from app.models import AppSetting
|
||
from app.vat import effective_vat_rate
|
||
|
||
app = make_app(tmp_path, "global-vat.db")
|
||
with app.app_context():
|
||
for key, value in {
|
||
"global_vat_override_enabled": "1",
|
||
"global_vat_override_name": "CPN",
|
||
"global_vat_override_rate": "8",
|
||
"global_vat_override_start": "2026-07-01",
|
||
"global_vat_override_end": "2026-07-31",
|
||
}.items():
|
||
db.session.merge(AppSetting(key=key, value=value))
|
||
db.session.commit()
|
||
rate = effective_vat_rate(Decimal("23"), date(2026, 7, 15))
|
||
result = calculate_costs(Decimal("108"), rate, 50)
|
||
assert rate == Decimal("8")
|
||
assert result["net"] == Decimal("100.00")
|
||
assert result["vat"] == Decimal("8.00")
|
||
assert result["deductible_vat"] == Decimal("4.00")
|
||
assert effective_vat_rate(Decimal("23"), date(2026, 8, 1)) == Decimal("23")
|
||
|
||
|
||
def test_only_admin_can_change_global_vat(tmp_path):
|
||
app = make_app(tmp_path, "global-vat-auth.db")
|
||
client = app.test_client()
|
||
client.post('/login', data={'email':'admin@example.com','password':'admin123!'})
|
||
response = client.post('/admin/global-vat', data={'enabled':'on','name':'CPN','rate':'8','start':'2026-07-01','end':'2026-07-31'}, headers={'X-Requested-With':'XMLHttpRequest'})
|
||
assert response.status_code == 200
|
||
assert response.json['ok'] is True
|
||
|
||
|
||
def test_api_bearer_token_authentication(tmp_path):
|
||
app = make_app(tmp_path, "bearer.db")
|
||
client = app.test_client()
|
||
|
||
token_response = client.post('/api/auth/token', data={
|
||
'username': 'admin@example.com',
|
||
'password': 'admin123!',
|
||
})
|
||
assert token_response.status_code == 200
|
||
token = token_response.json['access_token']
|
||
assert token_response.json['token_type'] == 'Bearer'
|
||
|
||
fresh_client = app.test_client()
|
||
response = fresh_client.get('/api/me', headers={
|
||
'Authorization': f'Bearer {token}',
|
||
})
|
||
assert response.status_code == 200
|
||
assert response.json['data']['email'] == 'admin@example.com'
|
||
|
||
|
||
def test_api_rejects_invalid_bearer_token(tmp_path):
|
||
app = make_app(tmp_path, "invalid-bearer.db")
|
||
response = app.test_client().get('/api/me', headers={
|
||
'Authorization': 'Bearer invalid-token',
|
||
})
|
||
assert response.status_code == 401
|
||
assert response.json['ok'] is False
|
||
|
||
|
||
def test_openapi_contains_password_flow(tmp_path):
|
||
app = make_app(tmp_path, "openapi-auth.db")
|
||
spec = app.test_client().get('/openapi.json').json
|
||
password_flow = spec['components']['securitySchemes']['oauth2Password']['flows']['password']
|
||
assert password_flow['tokenUrl'] == '/api/auth/token'
|
||
assert '/api/auth/token' in spec['paths']
|
||
|
||
|
||
def test_admin_can_delete_empty_company(tmp_path):
|
||
from app.extensions import db
|
||
from app.models import CompanySettings
|
||
|
||
app = make_app(tmp_path, "delete-company.db")
|
||
client = app.test_client()
|
||
client.post('/api/auth/login', data={'email':'admin@example.com','password':'admin123!'})
|
||
with app.app_context():
|
||
company = CompanySettings(name='Firma do usunięcia', region='mazowieckie')
|
||
db.session.add(company)
|
||
db.session.commit()
|
||
company_id = company.id
|
||
response = client.delete(f'/api/companies/{company_id}')
|
||
assert response.status_code == 200
|
||
assert response.json['ok'] is True
|
||
with app.app_context():
|
||
assert db.session.get(CompanySettings, company_id) is None
|
||
|
||
|
||
def test_company_with_users_cannot_be_deleted(tmp_path):
|
||
from app.extensions import db
|
||
from app.models import CompanySettings, User
|
||
|
||
app = make_app(tmp_path, "delete-company-blocked.db")
|
||
client = app.test_client()
|
||
client.post('/api/auth/login', data={'email':'admin@example.com','password':'admin123!'})
|
||
with app.app_context():
|
||
company = CompanySettings(name='Firma z użytkownikiem', region='mazowieckie')
|
||
db.session.add(company)
|
||
db.session.flush()
|
||
user = User(name='Pracownik', email='pracownik@example.com', role='user', company_id=company.id)
|
||
user.set_password('Haslo123!')
|
||
db.session.add(user)
|
||
db.session.commit()
|
||
company_id = company.id
|
||
response = client.delete(f'/api/companies/{company_id}')
|
||
assert response.status_code == 409
|
||
assert 'użytkowników' in response.json['message']
|
||
|
||
|
||
def test_company_settings_show_station_access_and_favorite_pagination(tmp_path):
|
||
app = make_app(tmp_path, "company-station-ui.db")
|
||
client = app.test_client()
|
||
client.post('/login', data={'email':'admin@example.com','password':'admin123!'})
|
||
html = client.get('/admin/companies').get_data(as_text=True)
|
||
assert 'Konfiguruj dostęp' in html
|
||
assert 'data-page-size="10"' in html
|
||
assert 'favorite-pagination' in html
|
||
assert 'Zapisz ulubione' in html
|
||
assert 'Usuń firmę' in html
|
||
|
||
def test_admin_stations_without_company_shows_company_picker(tmp_path):
|
||
app = make_app(tmp_path, "stations-picker.db")
|
||
client = app.test_client()
|
||
client.post('/login', data={'email':'admin@example.com','password':'admin123!'})
|
||
response = client.get('/stations')
|
||
html = response.get_data(as_text=True)
|
||
assert response.status_code == 200
|
||
assert 'Wybierz firmę' in html
|
||
assert 'name="company_id"' in html
|
||
assert 'Przejdź do konfiguracji' in html
|
||
|
||
|
||
def test_fuel_without_company_redirects_instead_of_500(tmp_path):
|
||
from app.extensions import db
|
||
from app.models import CompanySettings, User
|
||
|
||
app = make_app(tmp_path, "fuel-no-company.db")
|
||
client = app.test_client()
|
||
client.post('/login', data={'email':'admin@example.com','password':'admin123!'})
|
||
with app.app_context():
|
||
User.query.update({User.company_id: None})
|
||
CompanySettings.query.delete()
|
||
db.session.commit()
|
||
response = client.get('/fuel', follow_redirects=False)
|
||
assert response.status_code == 302
|
||
assert '/admin/companies' in response.headers['Location']
|
||
|
||
|
||
def test_delete_unused_fuel_card(tmp_path):
|
||
app=make_app(tmp_path, "delete_card.db")
|
||
client=app.test_client()
|
||
client.post('/login',data={'email':'admin@example.com','password':'admin123!'})
|
||
with app.app_context():
|
||
from app.extensions import db
|
||
from app.models import CompanySettings, FuelCard
|
||
company=CompanySettings.query.first()
|
||
card=FuelCard(name='Karta do usunięcia',provider='Test',company_id=company.id)
|
||
db.session.add(card);db.session.commit();card_id=card.id
|
||
response=client.delete(f'/api/fuel-cards/{card_id}')
|
||
assert response.status_code == 200
|
||
assert response.json['ok'] is True
|
||
with app.app_context():
|
||
from app.models import FuelCard
|
||
assert FuelCard.query.get(card_id) is None
|
||
|
||
|
||
def test_cannot_delete_assigned_fuel_card(tmp_path):
|
||
app=make_app(tmp_path, "assigned_card.db")
|
||
client=app.test_client()
|
||
client.post('/login',data={'email':'admin@example.com','password':'admin123!'})
|
||
with app.app_context():
|
||
from app.extensions import db
|
||
from app.models import CompanySettings, FuelCard, User
|
||
company=CompanySettings.query.first()
|
||
card=FuelCard(name='Karta przypisana',provider='Test',company_id=company.id)
|
||
db.session.add(card);db.session.flush()
|
||
user=User.query.filter_by(email='admin@example.com').first()
|
||
user.fuel_card_id=card.id
|
||
db.session.commit();card_id=card.id
|
||
response=client.delete(f'/api/fuel-cards/{card_id}')
|
||
assert response.status_code == 409
|
||
assert response.json['ok'] is False
|
||
assert 'użytkowników' in response.json['message']
|