160 lines
9.2 KiB
Python
160 lines
9.2 KiB
Python
"""Customer order browsing uses real login and server-side tenant grants."""
|
|
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from werkzeug.security import generate_password_hash
|
|
from logistics_chat import create_app, db, order_data
|
|
|
|
|
|
class CustomerOrderTests(unittest.TestCase):
|
|
def setUp(self):
|
|
folder = tempfile.TemporaryDirectory()
|
|
self.addCleanup(folder.cleanup)
|
|
root = Path(folder.name)
|
|
self.app = create_app({'TESTING': True, 'DATABASE': root/'history.db',
|
|
'UPLOAD_DIR': root/'uploads', 'BACKUP_DIR': root/'backups', 'LOG_DIR': root/'logs',
|
|
'DEEPSEEK_API_KEY': '', 'CUSTOMER_DEMO_ENABLED': False,
|
|
'ORDER_DATA_TENANT_ID': '', 'RATE_LIMIT_PER_MIN': 1000})
|
|
self.client = self.app.test_client()
|
|
self.client.post('/api/visitor/session', json={'key': '1'*64})
|
|
with self.app.app_context():
|
|
con = db.connect()
|
|
password = generate_password_hash('fixture-password')
|
|
for name, tenant in [('alice', 'a'), ('bob', 'b'), ('empty', '')]:
|
|
con.execute('INSERT INTO customer_users(username,password_hash,display_name,tenant_id) VALUES(?,?,?,?)',
|
|
(name, password, name, tenant))
|
|
order_data.import_records(con, [dict(tenant_id='a', order_no=f'A2026{i:04}',
|
|
inbound_code=f'IN2026{i:04}', status='PROCESSING', data_note='private-staff-note',
|
|
source_order_id='private-source-id') for i in range(25)] +
|
|
[dict(tenant_id='b', order_no='A20260024', inbound_code='IN20260024', status='DRAFT')])
|
|
self.foreign_id = con.execute("SELECT id FROM order_test_data WHERE tenant_id='b'").fetchone()[0]
|
|
con.commit()
|
|
con.close()
|
|
|
|
def login(self, name='alice'):
|
|
result = self.client.post('/api/customer/login', json={'username': name, 'password': 'fixture-password'})
|
|
self.assertEqual(result.status_code, 200)
|
|
|
|
def test_login_required_and_logout_revokes_browsing(self):
|
|
self.assertEqual(self.client.get('/api/customer/orders').status_code, 401)
|
|
self.login()
|
|
self.assertEqual(self.client.get('/api/customer/orders').status_code, 200)
|
|
self.client.post('/api/customer/logout')
|
|
for path in ['/api/customer/orders', '/api/customer/orders/1']:
|
|
response = self.client.get(path)
|
|
self.assertEqual(response.status_code, 401)
|
|
self.assertIn('no-store', response.headers['Cache-Control'])
|
|
|
|
def test_scope_cannot_be_forged_and_summary_is_minimal(self):
|
|
self.login()
|
|
response = self.client.get('/api/customer/orders?tenant_id=b', headers={'X-Tenant-Id': 'b'})
|
|
self.assertEqual(response.status_code, 200)
|
|
data = response.json
|
|
self.assertEqual(len(data['orders']), 20)
|
|
self.assertNotIn(self.foreign_id, [item['id'] for item in data['orders']])
|
|
for item in data['orders']:
|
|
self.assertEqual(set(item), {'id', 'order_no', 'inbound_code', 'status', 'country_code', 'warehouse_code', 'transport_mode'})
|
|
self.assertNotIn('private', response.get_data(as_text=True))
|
|
self.assertIn('no-store', response.headers['Cache-Control'])
|
|
self.assertEqual(self.client.get(f'/api/customer/orders/{self.foreign_id}?tenant_id=b').status_code, 404)
|
|
|
|
def test_cursor_paging_and_invalid_cursor(self):
|
|
self.login()
|
|
first = self.client.get('/api/customer/orders').json
|
|
second = self.client.get('/api/customer/orders', query_string={'before': first['next_before']}).json
|
|
self.assertEqual(len(second['orders']), 5)
|
|
self.assertIsNone(second['next_before'])
|
|
self.assertEqual(first['summary'], {'total': 25, 'pending': 0, 'processing': 25,
|
|
'completed': 0, 'cancelled': 0, 'other': 0})
|
|
self.assertEqual(second['summary'], first['summary'])
|
|
self.assertFalse({x['id'] for x in first['orders']} & {x['id'] for x in second['orders']})
|
|
for value in ['-1', '0', 'NaN', '9223372036854775808']:
|
|
self.assertEqual(self.client.get('/api/customer/orders', query_string={'before': value}).status_code, 400)
|
|
|
|
def test_detail_selects_exact_order_even_with_duplicate_identifiers(self):
|
|
self.login()
|
|
row = self.client.get('/api/customer/orders').json['orders'][0]
|
|
result = self.client.get('/api/customer/orders/'+str(row['id']))
|
|
self.assertEqual(result.status_code, 200)
|
|
self.assertEqual(result.json['card']['no'], row['order_no'])
|
|
self.assertNotIn('private', result.get_data(as_text=True))
|
|
self.assertIn('no-store', result.headers['Cache-Control'])
|
|
self.assertEqual(self.client.get('/api/customer/orders/99999').status_code, 404)
|
|
|
|
def test_empty_scope_and_configured_tenant_restriction(self):
|
|
self.login('empty')
|
|
self.assertEqual(self.client.get('/api/customer/orders').json['orders'], [])
|
|
self.assertEqual(self.client.get('/api/customer/orders').json['summary']['total'], 0)
|
|
self.assertEqual(self.client.get('/api/customer/orders/1').status_code, 404)
|
|
self.app.config['ORDER_DATA_TENANT_ID'] = 'b'
|
|
self.login()
|
|
self.assertEqual(self.client.get('/api/customer/orders').json['orders'], [])
|
|
self.assertEqual(self.client.get('/api/customer/orders').json['summary']['total'], 0)
|
|
self.assertEqual(self.client.get('/api/customer/orders/1').status_code, 404)
|
|
|
|
def test_multiple_grants_and_revocation(self):
|
|
with self.app.app_context():
|
|
con = db.connect()
|
|
con.execute("UPDATE customer_users SET tenant_id='' WHERE username='alice'")
|
|
con.execute("INSERT INTO customer_tenant_grants SELECT id,'a' FROM customer_users WHERE username='alice'")
|
|
con.execute("INSERT INTO customer_tenant_grants SELECT id,'b' FROM customer_users WHERE username='alice'")
|
|
con.commit()
|
|
con.close()
|
|
self.login()
|
|
self.assertEqual(self.client.get('/api/customer/orders').json['orders'][0]['id'], self.foreign_id)
|
|
self.assertEqual(self.client.get('/api/customer/orders').json['summary']['total'], 26)
|
|
with self.app.app_context():
|
|
con = db.connect()
|
|
con.execute("DELETE FROM customer_tenant_grants WHERE tenant_id='b'")
|
|
con.commit()
|
|
con.close()
|
|
self.assertEqual(self.client.get('/api/customer/orders').status_code, 401)
|
|
|
|
def test_unavailable_storage_has_recoverable_error(self):
|
|
self.login()
|
|
with patch('logistics_chat.customer_order_service.list_orders', side_effect=order_data.OrderDataUnavailable):
|
|
response = self.client.get('/api/customer/orders')
|
|
self.assertEqual(response.status_code, 503)
|
|
self.assertNotIn('sqlite', response.get_data(as_text=True))
|
|
|
|
def test_overview_counts_full_scope_and_filters_before_paging(self):
|
|
statuses = ['DRAFT', 'SUBMITTED', 'APPROVED', 'ACCEPTED', 'COMPLETED',
|
|
'CANCELLED', 'REJECTED', 'UNKNOWN_FUTURE_STATUS']
|
|
with self.app.app_context():
|
|
con = db.connect()
|
|
for index, status in enumerate(statuses):
|
|
con.execute('UPDATE order_test_data SET status=? WHERE tenant_id=? AND order_no=?',
|
|
(status, 'a', f'A2026{index:04}'))
|
|
con.commit()
|
|
con.close()
|
|
self.login()
|
|
expected = {'total': 25, 'pending': 4, 'processing': 17, 'completed': 1, 'cancelled': 1, 'other': 2}
|
|
all_data = self.client.get('/api/customer/orders?tenant_id=b').json
|
|
self.assertEqual(all_data['summary'], expected)
|
|
for status, count in [('pending', 4), ('processing', 17), ('completed', 1), ('cancelled', 1), ('other', 2)]:
|
|
with self.subTest(status=status):
|
|
response = self.client.get('/api/customer/orders', query_string={'status': status, 'tenant_id': 'b'})
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(len(response.json['orders']), count)
|
|
self.assertEqual(response.json['summary'], expected)
|
|
self.assertIsNone(response.json['next_before'])
|
|
self.assertNotIn(self.foreign_id, [item['id'] for item in response.json['orders']])
|
|
self.assertEqual({x['status'] for x in self.client.get('/api/customer/orders?status=other').json['orders']},
|
|
{'REJECTED', 'UNKNOWN_FUTURE_STATUS'})
|
|
for status in ['bad', 'PROCESSING', "pending' OR 1=1", '']:
|
|
self.assertEqual(self.client.get('/api/customer/orders', query_string={'status': status}).status_code, 400)
|
|
|
|
def test_filtered_cursor_keeps_summary_and_does_not_include_other_statuses(self):
|
|
self.login()
|
|
first = self.client.get('/api/customer/orders?status=processing').json
|
|
second = self.client.get('/api/customer/orders', query_string={'status': 'processing', 'before': first['next_before']}).json
|
|
self.assertEqual(len(first['orders']), 20)
|
|
self.assertEqual(len(second['orders']), 5)
|
|
self.assertEqual(second['summary']['processing'], 25)
|
|
empty = self.client.get('/api/customer/orders?status=completed').json
|
|
self.assertEqual(empty['orders'], [])
|
|
self.assertEqual(empty['summary']['total'], 25)
|