372 lines
19 KiB
Python
372 lines
19 KiB
Python
"""No e-conomic network traffic. DB tests are opt-in and roll back their rows."""
|
|
import asyncio
|
|
import os
|
|
from contextlib import contextmanager
|
|
from datetime import date, timedelta
|
|
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
|
|
from app.products.backend import economic_catalog as c
|
|
from app.products.backend import economic_documents as d
|
|
|
|
|
|
def product(**values):
|
|
return dict(id=12, name='Testvare', sales_price='100.00', sales_currency='DKK',
|
|
cost_price='40', cost_currency='DKK', economic_product_number='001-ABC',
|
|
economic_connection_id=1, economic_unit_number=1, status='active',
|
|
lifecycle_status='active', is_active_in_economic=True, **values)
|
|
|
|
|
|
def rule(**values):
|
|
base = dict(id=1, name='Rabat', kind='discount', value=10, currency='DKK', priority=0)
|
|
base.update(values)
|
|
return base
|
|
|
|
|
|
@pytest.mark.parametrize('value', ['00123', 'MS-365', 'a'*25])
|
|
def test_product_number_preserved(value):
|
|
assert c.product_number(value) == value
|
|
|
|
|
|
@pytest.mark.parametrize('value', [123, '', ' ', 'a'*26, None])
|
|
def test_product_number_rejected(value):
|
|
with pytest.raises(HTTPException):
|
|
c.product_number(value)
|
|
|
|
|
|
@pytest.mark.parametrize('value', ['NaN', 'Infinity', '-Infinity', None, 'abc'])
|
|
def test_decimal_rejects_non_finite(value):
|
|
with pytest.raises(HTTPException):
|
|
c.decimal(value)
|
|
|
|
|
|
def test_price_priority_and_zero():
|
|
rules = [rule(id=1, kind='fixed', value=80, category_id=4, priority=100),
|
|
rule(id=2, kind='fixed', value=70, product_id=12),
|
|
rule(id=3, kind='fixed', value=60, customer_id=2),
|
|
rule(id=4, kind='fixed', value=0, customer_id=2, product_id=12)]
|
|
assert c.calculate_price(product(category_id=4), rules, 2)['unit_price'] == '0.00'
|
|
assert c.calculate_price(product(), rules, 2, manual='12.345')['unit_price'] == '12.35'
|
|
|
|
|
|
def test_rule_conflict_not_resolved_by_row_order():
|
|
with pytest.raises(HTTPException, match='409'):
|
|
c.calculate_price(product(), [rule(), rule(id=2)])
|
|
|
|
|
|
def test_expired_rules_ignored():
|
|
assert c.calculate_price(product(), [rule(valid_to=date.today()-timedelta(days=1))])['unit_price'] == '100.00'
|
|
|
|
|
|
def test_cost_markup_is_not_margin():
|
|
assert c.calculate_price(product(), [rule(kind='cost_markup', value=25)])['unit_price'] == '50.00'
|
|
|
|
|
|
def test_unknown_cost_currency_blocks_markup():
|
|
p = product()
|
|
p['cost_currency'] = None
|
|
with pytest.raises(HTTPException):
|
|
c.calculate_price(p, [rule(kind='cost_markup')])
|
|
|
|
|
|
def test_reference_price_fallback_keeps_currency():
|
|
p = product()
|
|
p.update(sales_price=None, economic_sales_price_reference='22', economic_currency='EUR')
|
|
assert c.calculate_price(p, [], currency='EUR')['unit_price'] == '22.00'
|
|
with pytest.raises(HTTPException):
|
|
c.calculate_price(p, [], currency='DKK')
|
|
|
|
|
|
def test_snapshot_preserves_saved_price_and_zero():
|
|
snap = d.line_snapshot({'quantity': 2, 'unit_price': 0}, product(), 1, 'DKK')
|
|
assert snap['unit_price'] == '0.00'
|
|
assert snap['economic_product_number'] == '001-ABC'
|
|
assert snap['net_total'] == '0.00'
|
|
|
|
|
|
@pytest.mark.parametrize('changes', [{'economic_product_number': None}, {'economic_connection_id': 2},
|
|
{'is_active_in_economic': False}, {'deleted_at': 'today'}, {'status': 'inactive'}])
|
|
def test_export_invalid_product_blocks(changes):
|
|
p = product()
|
|
p.update(changes)
|
|
with pytest.raises(HTTPException):
|
|
d.line_snapshot({'quantity':1,'unit_price':10}, p, 1, 'DKK')
|
|
|
|
|
|
def test_export_does_not_relabel_currency():
|
|
with pytest.raises(HTTPException):
|
|
d.line_snapshot({'quantity':1,'unit_price':10,'currency':'EUR'}, product(), 1, 'DKK')
|
|
|
|
|
|
@pytest.mark.parametrize('flag', ['ECONOMIC_READ_ONLY', 'ECONOMIC_DRY_RUN'])
|
|
@pytest.mark.parametrize('method', ['POST', 'PUT', 'PATCH', 'DELETE'])
|
|
def test_safety_prevents_network_for_every_write(monkeypatch, flag, method):
|
|
monkeypatch.setattr(c.settings, flag, True)
|
|
with pytest.raises(c.RemoteError) as error:
|
|
asyncio.run(c.EconomicClient().request(method, 'products', {}))
|
|
assert error.value.status == 423
|
|
|
|
|
|
@pytest.mark.parametrize('method,path', [('PUT','products/1'), ('PATCH','products/1'), ('DELETE','products/1'),
|
|
('POST','invoices/drafts'), ('POST','suppliers'), ('POST','journals/1/vouchers')])
|
|
def test_economic_write_policy_blocks_everything_except_create_customer_product_order(monkeypatch, method, path):
|
|
monkeypatch.setattr(c.settings, 'ECONOMIC_READ_ONLY', False)
|
|
monkeypatch.setattr(c.settings, 'ECONOMIC_DRY_RUN', False)
|
|
with pytest.raises(c.RemoteError) as error:
|
|
asyncio.run(c.EconomicClient().request(method, path, {}))
|
|
assert error.value.status == 403
|
|
|
|
|
|
@pytest.mark.parametrize('path', ['customers', 'products', 'orders/drafts'])
|
|
def test_economic_write_policy_allows_only_three_create_operations(path):
|
|
from app.core.economic_write_policy import assert_economic_write_allowed
|
|
assert_economic_write_allowed('POST', path)
|
|
|
|
|
|
def test_four_eyes_context_allows_only_customer_and_product_put():
|
|
from app.core.economic_write_policy import approved_four_eyes_write, assert_economic_write_allowed
|
|
with pytest.raises(HTTPException):
|
|
assert_economic_write_allowed('PUT', 'customers/42')
|
|
with approved_four_eyes_write():
|
|
assert_economic_write_allowed('PUT', 'customers/42')
|
|
assert_economic_write_allowed('PUT', 'products/ABC-1')
|
|
with pytest.raises(HTTPException):
|
|
assert_economic_write_allowed('PUT', 'orders/drafts/12')
|
|
with pytest.raises(HTTPException):
|
|
assert_economic_write_allowed('DELETE', 'products/ABC-1')
|
|
with pytest.raises(HTTPException):
|
|
assert_economic_write_allowed('PUT', 'products/ABC-1')
|
|
|
|
|
|
def test_confirmed_order_context_allows_only_numeric_draft_put():
|
|
from app.core.economic_write_policy import approved_order_draft_update, assert_economic_write_allowed
|
|
with pytest.raises(HTTPException):
|
|
assert_economic_write_allowed('PUT', 'orders/drafts/12')
|
|
with approved_order_draft_update():
|
|
assert_economic_write_allowed('PUT', 'orders/drafts/12')
|
|
with pytest.raises(HTTPException):
|
|
assert_economic_write_allowed('PUT', 'orders/12')
|
|
with pytest.raises(HTTPException):
|
|
assert_economic_write_allowed('DELETE', 'orders/drafts/12')
|
|
|
|
|
|
def test_order_update_payload_strips_read_only_fields_and_keeps_product():
|
|
from app.modules.orders.backend.economic_sync import build_update_payload
|
|
payload, _ = build_update_payload({
|
|
'date':'2026-09-13','currency':'DKK','customer':{'customerNumber':1},
|
|
'orderNumber':42,'netAmount':100,'lines':[{'lineNumber':1,'product':{'productNumber':'P1'},'quantity':1,'unitNetPrice':10}],
|
|
}, [{'description':'Ny','quantity':2,'unit_price':25,'discount_percentage':10}], 'Note')
|
|
assert 'orderNumber' not in payload and 'netAmount' not in payload
|
|
assert payload['lines'] == [{'product':{'productNumber':'P1'},'description':'Ny','quantity':2.0,'unitNetPrice':25.0,'discountPercentage':10.0}]
|
|
|
|
|
|
def test_four_eyes_change_field_allowlist():
|
|
from app.products.backend.economic_routes import _validate_change_fields
|
|
assert _validate_change_fields('product', {'name': 'Nyt navn'}) == {'name': 'Nyt navn'}
|
|
assert _validate_change_fields('customer', {'city': 'Værløse'}) == {'city': 'Værløse'}
|
|
with pytest.raises(HTTPException):
|
|
_validate_change_fields('product', {'productNumber': 'NYT'})
|
|
with pytest.raises(HTTPException):
|
|
_validate_change_fields('customer', {'customerNumber': 99})
|
|
|
|
|
|
def test_collection_paginates_and_rejects_loop(monkeypatch):
|
|
client = c.EconomicClient()
|
|
client.request = AsyncMock(side_effect=[{'collection':[1], 'pagination':{'nextPage':'products?skip=1'}}, {'collection':[2]}])
|
|
assert asyncio.run(client.collection('products')) == [1,2]
|
|
client.request = AsyncMock(return_value={'collection':[], 'pagination':{'nextPage':'products?pagesize=1000'}})
|
|
with pytest.raises(c.RemoteError):
|
|
asyncio.run(client.collection('products'))
|
|
|
|
|
|
def test_verify_uses_actual_self_schema(monkeypatch):
|
|
monkeypatch.setattr(c, 'query', lambda *a, **k: {'agreement_number':'123','currency':'DKK'})
|
|
client = c.EconomicClient()
|
|
client.request = AsyncMock(return_value={'agreementNumber':123,'settings':{'baseCurrency':'DKK'}})
|
|
assert asyncio.run(client.verify(1))['agreement_number'] == '123'
|
|
client.request = AsyncMock(return_value={'agreementNumber':999,'settings':{'baseCurrency':'DKK'}})
|
|
with pytest.raises(c.RemoteError):
|
|
asyncio.run(client.verify(1))
|
|
|
|
|
|
def test_export_duplicate_never_posts(monkeypatch):
|
|
monkeypatch.setattr(d, 'query', lambda *a, **kw: {'status':'verified','kind':'order','economic_number':'42','id':'abc'})
|
|
preflight = AsyncMock(side_effect=AssertionError('must not preflight or POST again'))
|
|
monkeypatch.setattr(d, 'preflight', preflight)
|
|
result = asyncio.run(d.export_document({'id':1}, 'order', 'same', 1, []))
|
|
assert result['economic_order_number'] == '42'
|
|
preflight.assert_not_called()
|
|
|
|
|
|
def test_uncertain_export_blocks_retry(monkeypatch):
|
|
monkeypatch.setattr(d, 'query', lambda *a, **kw: {'status':'uncertain'})
|
|
with pytest.raises(HTTPException):
|
|
asyncio.run(d.export_document({'id':1}, 'order', 'same', 1, []))
|
|
|
|
|
|
@pytest.fixture
|
|
def catalog_db(monkeypatch):
|
|
if os.environ.get('BMC_CATALOG_DB_TESTS') != '1':
|
|
pytest.skip('Opt-in local transactional database tests')
|
|
import psycopg2
|
|
from psycopg2.extras import RealDictCursor
|
|
conn = psycopg2.connect(c.settings.DATABASE_URL)
|
|
@contextmanager
|
|
def tx():
|
|
with conn.cursor(cursor_factory=RealDictCursor) as cur:
|
|
cur.execute('SAVEPOINT catalog_test')
|
|
try:
|
|
yield cur
|
|
cur.execute('RELEASE SAVEPOINT catalog_test')
|
|
except Exception:
|
|
cur.execute('ROLLBACK TO SAVEPOINT catalog_test')
|
|
raise
|
|
monkeypatch.setattr(c, 'transaction', tx)
|
|
monkeypatch.setattr(d, 'transaction', tx)
|
|
try:
|
|
yield
|
|
finally:
|
|
conn.rollback()
|
|
conn.close()
|
|
|
|
|
|
def test_database_identity_outbox_import_and_rollback(catalog_db):
|
|
import psycopg2
|
|
connection = c.query("INSERT INTO economic_catalog_connections(agreement_number,name,currency) VALUES('catalog-test-'||gen_random_uuid(),'TEST','DKK') RETURNING *", one=True)
|
|
remote = {'productNumber':'00123','name':'Original','salesPrice':100,'productGroup':{'productGroupNumber':1},'unit':{'unitNumber':1}}
|
|
assert c.import_product(connection, remote) == 'created'
|
|
assert c.import_product(connection, remote) == 'unchanged'
|
|
p = c.query('SELECT * FROM products WHERE economic_connection_id=%s', (connection['id'],), one=True)
|
|
with pytest.raises(psycopg2.Error):
|
|
c.query("UPDATE products SET economic_product_number='changed' WHERE id=%s", (p['id'],))
|
|
c.query('UPDATE products SET name=%s WHERE id=%s', ('Hub edited', p['id']))
|
|
jobs = c.query("SELECT * FROM economic_catalog_jobs WHERE product_id=%s AND kind='name'", (p['id'],))
|
|
assert jobs == []
|
|
c.import_product(connection, remote)
|
|
current = c.query('SELECT * FROM products WHERE id=%s', (p['id'],), one=True)
|
|
assert current['name'] == 'Hub edited' and current['economic_sync_status'] == 'local_only'
|
|
c.import_product(connection, {**remote, 'name':'External edit'})
|
|
assert c.query('SELECT economic_sync_status FROM products WHERE id=%s', (p['id'],), one=True)['economic_sync_status'] == 'conflict'
|
|
|
|
|
|
def test_database_create_double_click_same_reservation(catalog_db):
|
|
connection = c.query("INSERT INTO economic_catalog_connections(agreement_number,name,currency) VALUES('catalog-test-'||gen_random_uuid(),'TEST','DKK') RETURNING *", one=True)
|
|
p = c.query("INSERT INTO products(name) VALUES('TEST') RETURNING id", one=True)
|
|
first = c.enqueue(connection['id'], 'create', None, p['id'], {'productGroup':{'productGroupNumber':1}})
|
|
second = c.enqueue(connection['id'], 'create', None, p['id'], {})
|
|
assert first == second
|
|
assert c.query('SELECT next_number FROM economic_catalog_connections WHERE id=%s', (connection['id'],), one=True)['next_number'] == 2
|
|
|
|
|
|
def test_database_timeout_is_uncertain_and_retry_cannot_post(catalog_db, monkeypatch):
|
|
connection = c.query("INSERT INTO economic_catalog_connections(agreement_number,name,currency) VALUES('catalog-test-'||gen_random_uuid(),'TEST','DKK') RETURNING *", one=True)
|
|
monkeypatch.setattr(d, 'preflight', AsyncMock(return_value={'payload':{},'lines':[]}))
|
|
request = AsyncMock(side_effect=asyncio.TimeoutError())
|
|
monkeypatch.setattr(c.EconomicClient, 'request', request)
|
|
with pytest.raises(HTTPException):
|
|
asyncio.run(d.export_document(connection,'order','timeout-test',1,[]))
|
|
row = c.query('SELECT * FROM economic_document_exports WHERE connection_id=%s', (connection['id'],), one=True)
|
|
assert row['status'] == 'uncertain'
|
|
with pytest.raises(HTTPException):
|
|
asyncio.run(d.export_document(connection,'order','timeout-test',1,[]))
|
|
assert request.call_count == 1
|
|
|
|
|
|
def test_database_local_name_change_does_not_queue_external_write(catalog_db, monkeypatch):
|
|
connection = c.query("INSERT INTO economic_catalog_connections(agreement_number,name,currency,enabled) VALUES('catalog-test-'||gen_random_uuid(),'TEST','DKK',true) RETURNING *", one=True)
|
|
c.import_product(connection, {'productNumber':'TEST-001','name':'Original','productGroup':{'productGroupNumber':1}})
|
|
p = c.query('SELECT * FROM products WHERE economic_connection_id=%s', (connection['id'],), one=True)
|
|
c.query("UPDATE products SET name='First' WHERE id=%s", (p['id'],))
|
|
c.query("UPDATE products SET name='Second' WHERE id=%s", (p['id'],))
|
|
assert not c.query("SELECT id FROM economic_catalog_jobs WHERE product_id=%s AND kind='name'", (p['id'],))
|
|
current = c.query('SELECT name,economic_sync_status FROM products WHERE id=%s', (p['id'],), one=True)
|
|
assert current == {'name':'Second','economic_sync_status':'local_only'}
|
|
|
|
|
|
def test_database_create_timeout_retains_reserved_number(catalog_db, monkeypatch):
|
|
connection = c.query("INSERT INTO economic_catalog_connections(agreement_number,name,currency,enabled) VALUES('catalog-test-'||gen_random_uuid(),'TEST','DKK',true) RETURNING *", one=True)
|
|
p = c.query("INSERT INTO products(name) VALUES('New test product') RETURNING id", one=True)
|
|
c.query("INSERT INTO economic_catalog_references(connection_id,kind,number,name,payload) VALUES(%s,'product-groups',1,'Test','{}')", (connection['id'],))
|
|
enqueued = c.enqueue(connection['id'], 'create', None, p['id'], {'productGroup':{'productGroupNumber':1}})
|
|
job = c.query('SELECT * FROM economic_catalog_jobs WHERE id=%s', (enqueued['id'],), one=True)
|
|
monkeypatch.setattr(c.EconomicClient, 'verify', AsyncMock(return_value=connection))
|
|
async def remote(method, path, payload=None):
|
|
if path.startswith('products/'):
|
|
raise c.RemoteError(404, 'not found')
|
|
if method == 'POST':
|
|
raise asyncio.TimeoutError()
|
|
return {'productGroupNumber':1}
|
|
monkeypatch.setattr(c.EconomicClient, 'request', staticmethod(remote))
|
|
with pytest.raises(asyncio.TimeoutError):
|
|
asyncio.run(c.process_job(job))
|
|
assert c.query('SELECT status FROM economic_catalog_jobs WHERE id=%s', (job['id'],), one=True)['status'] == 'uncertain'
|
|
assert c.enqueue(connection['id'], 'create', None, p['id'], {}) == enqueued
|
|
|
|
|
|
def test_database_paging_checkpoint_survives_interrupted_import(catalog_db, monkeypatch):
|
|
connection = c.query("INSERT INTO economic_catalog_connections(agreement_number,name,currency) VALUES('catalog-test-'||gen_random_uuid(),'TEST','DKK') RETURNING *", one=True)
|
|
enqueued = c.enqueue(connection['id'], 'import', None)
|
|
job = c.query('SELECT * FROM economic_catalog_jobs WHERE id=%s', (enqueued['id'],), one=True)
|
|
monkeypatch.setattr(c.EconomicClient, 'verify', AsyncMock(return_value=connection))
|
|
seen = []
|
|
fail = True
|
|
async def remote(method, path, payload=None):
|
|
seen.append(path)
|
|
if path == 'products?pagesize=1000':
|
|
return {'collection':[{'productNumber':'TEST-01','name':'Test','productGroup':{'productGroupNumber':1}}], 'pagination':{'nextPage':'products?skip=1'}}
|
|
if path == 'products?skip=1' and fail:
|
|
raise c.RemoteError(503, 'interrupted')
|
|
return {'collection':[]}
|
|
monkeypatch.setattr(c.EconomicClient, 'request', staticmethod(remote))
|
|
with pytest.raises(c.RemoteError):
|
|
asyncio.run(c.process_job(job))
|
|
assert not c.query('SELECT id FROM products WHERE economic_connection_id=%s', (connection['id'],))
|
|
fail = False
|
|
result = asyncio.run(c.process_job(job))
|
|
assert result['created'] == 1
|
|
assert seen.count('products?pagesize=1000') == 1
|
|
|
|
|
|
def test_manual_price_requires_permission_but_saved_contract_is_preserved(monkeypatch):
|
|
from app.products.backend import economic_pricing as pricing
|
|
monkeypatch.setattr(pricing, 'query', lambda sql,*a,**k: [] if 'economic_price_rules' in sql else product())
|
|
line = {'line_key':'a','product_id':12,'unit_price':55,'discount_percentage':0}
|
|
with pytest.raises(HTTPException) as exc:
|
|
pricing.validate_manual_prices([line], 2, False)
|
|
assert exc.value.status_code == 403
|
|
pricing.validate_manual_prices([line], 2, False, [line])
|
|
pricing.validate_manual_prices([line], 2, True)
|
|
|
|
|
|
def test_cost_filter_covers_nested_response_and_history():
|
|
from app.products.backend.economic_access import without_cost
|
|
result = without_cost({'cost_price':25, 'name':'Test', 'sales_price':100,
|
|
'economic_snapshot':{'costPrice':40}, 'nested':{'unitCostPrice':20},
|
|
'history':[{'price_type':'supplier_price','old_price':10,'new_price':20}, {'price_type':'sales_price','new_price':100}]})
|
|
assert result == {'name':'Test','sales_price':100,'nested':{},'history':[{'price_type':'sales_price','new_price':100}]}
|
|
|
|
|
|
def test_catalog_api_requires_authentication():
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
from app.products.backend.economic_routes import router
|
|
app = FastAPI()
|
|
app.include_router(router)
|
|
assert TestClient(app).get('/economic/catalog/status').status_code == 401
|
|
assert TestClient(app).post('/products/1/economic-create', json={'connection_id':1,'group_number':1}).status_code == 401
|
|
|
|
|
|
def test_catalog_reader_cannot_create_product(monkeypatch):
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
from app.core.auth_dependencies import get_current_user
|
|
from app.core.auth_service import AuthService
|
|
from app.products.backend.economic_routes import router
|
|
app = FastAPI()
|
|
app.include_router(router)
|
|
app.dependency_overrides[get_current_user] = lambda: {'id':1,'username':'reader','is_superadmin':False}
|
|
monkeypatch.setattr(AuthService, 'user_has_permission', lambda uid, permission: permission == 'economic.catalog.view')
|
|
assert TestClient(app).post('/products/1/economic-create', json={'connection_id':1,'group_number':1}).status_code == 403
|