"""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)