115 lines
4.8 KiB
Python
115 lines
4.8 KiB
Python
"""Debug: check accept task response body."""
|
|
import sys, os
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
os.environ['APP_ENV'] = 'test'
|
|
os.environ['MYSQL_DATABASE'] = ':memory:'
|
|
os.environ['SECRET_KEY'] = 'test-secret-key-for-testing-only'
|
|
os.environ['JWT_EXPIRE_MINUTES'] = '60'
|
|
os.environ['AI_PROVIDER'] = 'mock'
|
|
os.environ['AUTH_TOKEN_STORE_PATH'] = 'D:/tmp/order-flow-test/revoked_tokens.json'
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
from backend.app.db import Base
|
|
from backend.app.core.security import create_access_token, hash_password
|
|
from backend.app.main import app
|
|
from backend.app.api.deps import get_db_session
|
|
from backend.tests.conftest import SyncTestClient
|
|
from backend.app.models.system import Role, User, Menu, RoleMenu
|
|
from backend.app.models.business import Customer, Product, Supplier, SalesOrder, LogisticsTask
|
|
|
|
engine = create_engine('sqlite:///:memory:', echo=False, connect_args={'check_same_thread': False})
|
|
Base.metadata.create_all(engine)
|
|
connection = engine.connect()
|
|
transaction = connection.begin()
|
|
session = Session(bind=connection)
|
|
|
|
roles = [
|
|
Role(id=1, role_name='Admin', role_code='admin', status=1),
|
|
Role(id=4, role_name='Driver', role_code='driver', status=1),
|
|
]
|
|
session.add_all(roles)
|
|
session.flush()
|
|
|
|
users = [
|
|
User(id=1, username='admin01', password_hash=hash_password('admin123'), real_name='Admin', mobile='13800000001', role_id=1, status=1),
|
|
User(id=5, username='driver01', password_hash=hash_password('driver123'), real_name='Driver', mobile='13800000005', role_id=4, status=1),
|
|
]
|
|
session.add_all(users)
|
|
session.flush()
|
|
|
|
menus = [
|
|
Menu(id=8, menu_name='Driver', menu_path='/driver', menu_type='page', permission_code='driver:task:list', sort_no=6, status=1),
|
|
Menu(id=9, menu_name='Accept', menu_path='', menu_type='button', permission_code='driver:task:accept', parent_id=8, sort_no=1, status=1),
|
|
Menu(id=25, menu_name='TaskList', menu_path='', menu_type='button', permission_code='logistics:task:list', parent_id=8, sort_no=4, status=1),
|
|
Menu(id=26, menu_name='TaskCreate', menu_path='', menu_type='button', permission_code='logistics:task:create', parent_id=8, sort_no=5, status=1),
|
|
]
|
|
session.add_all(menus)
|
|
session.flush()
|
|
|
|
role_menus = [
|
|
*[RoleMenu(role_id=1, menu_id=m.id) for m in menus],
|
|
RoleMenu(role_id=4, menu_id=8), RoleMenu(role_id=4, menu_id=9),
|
|
]
|
|
session.add_all(role_menus)
|
|
session.flush()
|
|
|
|
def override_get_db():
|
|
try:
|
|
yield session
|
|
finally:
|
|
pass
|
|
|
|
app.dependency_overrides[get_db_session] = override_get_db
|
|
|
|
admin_h = {'Authorization': f'Bearer {create_access_token({"user_id": 1, "role_code": "admin", "username": "admin01"})}'}
|
|
driver_h = {'Authorization': f'Bearer {create_access_token({"user_id": 5, "role_code": "driver", "username": "driver01"})}'}
|
|
|
|
with SyncTestClient(app, raise_server_exceptions=False) as c:
|
|
customer = Customer(customer_name='Test', mobile='13800001111', settlement_type='immediate', settlement_days=0, salesman_id=2, credit_limit=0, deleted=0)
|
|
product = Product(product_name='TestProd', specification='100x100', unit='pcs', category='cat', cost_price=60.0, sale_price=100.0, status=1)
|
|
supplier = Supplier(supplier_name='TestFactory', supplier_type='factory', contact_name='Contact', contact_mobile='13900001111', status=1)
|
|
session.add_all([customer, product, supplier])
|
|
session.flush()
|
|
|
|
order = SalesOrder(order_no='TEST-001', customer_id=customer.id, customer_name='Test',
|
|
customer_mobile='13800001111', salesman_id=2, order_status='approved',
|
|
factory_id=supplier.id, sale_price_total=1000.0, cost_price_total=600.0,
|
|
profit_total=400.0, profit_rate=40.0, deleted=0)
|
|
session.add(order)
|
|
session.flush()
|
|
|
|
# Create task
|
|
resp = c.post('/api/logistics/tasks', headers=admin_h, json={
|
|
'order_id': order.id, 'driver_id': 5,
|
|
'pickup_address': 'F', 'delivery_address': 'W',
|
|
'pickup_content': 'G', 'quantity': 10,
|
|
})
|
|
print(f'Create task: {resp.status_code}, body: {resp.json()}')
|
|
task_id = resp.json()['data']['task_id']
|
|
|
|
# Check task in DB
|
|
session.expire_all()
|
|
task = session.get(LogisticsTask, task_id)
|
|
print(f'Task in DB: status={task.status}, order_id={task.order_id}')
|
|
|
|
# Accept task
|
|
resp = c.post(f'/api/driver/tasks/{task_id}/accept', headers=driver_h, json={})
|
|
print(f'Accept: status={resp.status_code}')
|
|
print(f'Accept body: {resp.json()}')
|
|
|
|
# Check task in DB
|
|
session.expire_all()
|
|
task = session.get(LogisticsTask, task_id)
|
|
print(f'Task after accept: status={task.status}')
|
|
|
|
# Check order in DB
|
|
order = session.get(SalesOrder, order.id)
|
|
print(f'Order after accept: status={order.order_status}')
|
|
|
|
app.dependency_overrides.clear()
|
|
transaction.rollback()
|
|
session.close()
|
|
connection.close()
|