142 lines
6.9 KiB
Python
142 lines
6.9 KiB
Python
from app_new import db
|
|
from app_new.core.models.college import Building, Room, Site
|
|
from app_new.core.models.equipment import Equipment, EquipmentCategory, EquipmentLifecycleEvent, EquipmentQuantityMovement
|
|
from app_new.core.models.maintenance import Lot, LotTask
|
|
from app_new.core.models.planning import ScheduledTask
|
|
from app_new.core.services.equipment_lifecycle import complete_repair as complete_lifecycle_repair, propose_disposal, start_repair as start_lifecycle_repair
|
|
from app_new.core.models.user import User
|
|
from app_new.equipments.main import complete_repair, move_to_room, start_repair
|
|
from flask_login import login_user
|
|
|
|
|
|
def _rooms(app):
|
|
with app.app_context():
|
|
building = Building(name='Bâtiment transfert')
|
|
db.session.add(building)
|
|
db.session.flush()
|
|
source = Room(name='Salle source', building_id=building.id)
|
|
target = Room(name='Salle cible', building_id=building.id)
|
|
db.session.add_all([source, target])
|
|
db.session.commit()
|
|
return source.id, target.id
|
|
|
|
|
|
def test_partial_group_transfer_updates_both_rooms(authenticated_client, app):
|
|
source_room_id, target_room_id = _rooms(app)
|
|
with app.app_context():
|
|
root = Equipment(name='Chaises', is_group=True, quantity=30, mobility='mobile')
|
|
db.session.add(root)
|
|
db.session.flush()
|
|
source = Equipment(
|
|
name='Chaises', parent_id=root.id, room_id=source_room_id,
|
|
is_group=True, quantity=10, mobility='mobile',
|
|
)
|
|
db.session.add(source)
|
|
db.session.commit()
|
|
source_id, root_id = source.id, root.id
|
|
assert db.session.get(Room, target_room_id) is not None
|
|
assert db.session.get(Equipment, source_id) is not None
|
|
|
|
admin = User.query.filter_by(role='admin').first()
|
|
with app.test_request_context(method='POST', data={'room_id': target_room_id, 'quantity': 4, 'reason': 'Réorganisation'}):
|
|
login_user(admin)
|
|
response = move_to_room.__wrapped__(source_id)
|
|
assert response.status_code == 302
|
|
|
|
with app.app_context():
|
|
source = db.session.get(Equipment, source_id)
|
|
target = Equipment.query.filter_by(parent_id=root_id, room_id=target_room_id).one()
|
|
movement = EquipmentQuantityMovement.query.filter_by(root_equipment_id=root_id).one()
|
|
assert source.quantity == 6
|
|
assert target.quantity == 4
|
|
assert movement.quantity == 4
|
|
|
|
|
|
def test_fixed_equipment_cannot_be_moved(authenticated_client, app):
|
|
source_room_id, target_room_id = _rooms(app)
|
|
with app.app_context():
|
|
equipment = Equipment(name='Luminaire fixe', room_id=source_room_id, mobility='fixe')
|
|
db.session.add(equipment)
|
|
db.session.commit()
|
|
equipment_id = equipment.id
|
|
admin = User.query.filter_by(role='admin').first()
|
|
with app.test_request_context(method='POST', data={'room_id': target_room_id}):
|
|
login_user(admin)
|
|
response = move_to_room.__wrapped__(equipment_id)
|
|
assert response.status_code == 302
|
|
with app.app_context():
|
|
assert db.session.get(Equipment, equipment_id).room_id == source_room_id
|
|
|
|
|
|
def test_repair_workflow(authenticated_client, app):
|
|
with app.app_context():
|
|
equipment = Equipment(name='Photocopieur', mobility='mobile')
|
|
db.session.add(equipment)
|
|
db.session.commit()
|
|
equipment_id = equipment.id
|
|
admin = User.query.filter_by(role='admin').first()
|
|
with app.test_request_context(method='POST'):
|
|
login_user(admin)
|
|
assert start_repair.__wrapped__(equipment_id).status_code == 302
|
|
assert db.session.get(Equipment, equipment_id).lifecycle_status == 'en_reparation'
|
|
with app.test_request_context(method='POST'):
|
|
login_user(admin)
|
|
assert complete_repair.__wrapped__(equipment_id).status_code == 302
|
|
equipment = db.session.get(Equipment, equipment_id)
|
|
assert equipment.lifecycle_status == 'en_service'
|
|
assert equipment.unavailable_since is None
|
|
|
|
|
|
def test_site_can_represent_staff_housing(authenticated_client, app):
|
|
response = authenticated_client.post('/equipments/buildings/sites', data={
|
|
'name': 'Logements de fonction', 'site_type': 'logements',
|
|
})
|
|
assert response.status_code == 302
|
|
with app.app_context():
|
|
assert Site.query.filter_by(name='Logements de fonction', site_type='logements').one()
|
|
|
|
|
|
def test_quantitative_breakage_is_individualized_and_suspends_preventive(app):
|
|
with app.app_context():
|
|
building = Building(name='Bâtiment cycle préventif')
|
|
db.session.add(building); db.session.flush()
|
|
room = Room(name='B02 cycle préventif', building_id=building.id)
|
|
category = EquipmentCategory(name='Mobilier cycle préventif')
|
|
lot = Lot(name='Lot mobilier cycle préventif', category=category)
|
|
root = Equipment(name='Chaises cycle préventif', category=category, lot=lot,
|
|
is_group=True, quantity=30, mobility='mobile', management_mode='quantitatif')
|
|
db.session.add_all([room, category, lot, root]); db.session.flush()
|
|
room_group = Equipment(name=root.name, parent_id=root.id, room_id=room.id,
|
|
is_group=True, quantity=30, mobility='mobile')
|
|
rule = LotTask(lot=lot, tache='Contrôle annuel des chaises', trigger_type='calendar',
|
|
application_scope='group', unavailable_policy='suspend', is_active=True)
|
|
db.session.add_all([room_group, rule]); db.session.commit()
|
|
|
|
units, intervention = start_lifecycle_repair(room_group, quantity=2, reason='Pieds cassés')
|
|
for unit in units:
|
|
unit.create_scheduled_tasks_from_lot(commit=False)
|
|
db.session.commit()
|
|
|
|
assert room_group.unindividualized_quantity == 28
|
|
assert intervention.quantity_affected == 2
|
|
assert all(unit.lifecycle_status == 'en_reparation' for unit in units)
|
|
assert ScheduledTask.query.filter(
|
|
ScheduledTask.equipment_id.in_([unit.id for unit in units]),
|
|
ScheduledTask.status == 'suspended',
|
|
).count() == 2
|
|
assert EquipmentLifecycleEvent.query.filter_by(event_type='retrait_reparation').count() >= 2
|
|
|
|
complete_lifecycle_repair(units[0], reason='Chaise réparée')
|
|
propose_disposal(units[1], reason='Pied irréparable')
|
|
db.session.commit()
|
|
assert units[0].lifecycle_status == 'en_service'
|
|
assert units[0].scheduled_tasks.filter_by(status='planned').count() == 1
|
|
assert units[1].lifecycle_status == 'mise_au_rebut_planifiee'
|
|
assert units[1].scheduled_tasks.filter_by(status='cancelled').count() == 1
|
|
|
|
|
|
def test_lifecycle_and_lot_preventive_pages_render(authenticated_client, app):
|
|
assert authenticated_client.get('/equipments/lifecycle').status_code == 200
|
|
with app.app_context():
|
|
assert app.jinja_env.get_template('lots/detail.html')
|
|
assert app.jinja_env.get_template('lots/task_form.html')
|