"""Admin break-glass over policy gating, and the holds that make it stick.

Policy gating is right almost always and wrong at 3am. These tests pin the
override's boundaries: it must be explicit, admin-only, reasoned, audited, and
it must NOT become a way to acquire an elevated permission the caller lacks.

The hold half exists because an override on a desired-mode action is a fight
with the reconciler - without suppression the next check-in re-enqueues the
opposite command and the override looks broken.
"""

from datetime import datetime, timedelta

from werkzeug.security import generate_password_hash

from shopdb.core.models import Setting


def _enable_policies(db):
    if not Setting.query.filter_by(key='computers_configpolicies_enabled').first():
        db.session.add(Setting(key='computers_configpolicies_enabled',
                               value='true'))
        db.session.commit()


def _endpoint(db, hostname='holdbox'):
    """An Asset + Computer pair, the minimum a policy can resolve against."""
    from shopdb.api import Asset, AssetType
    from plugins.computers.models import Computer

    assettype = AssetType.query.filter_by(assettype='computer').first()
    if assettype is None:
        assettype = AssetType(assettype='computer', pluginname='computers',
                              tablename='computers')
        db.session.add(assettype)
        db.session.flush()

    asset = Asset(assetnumber=hostname, name=hostname,
                  assettypeid=assettype.assettypeid, isactive=True)
    db.session.add(asset)
    db.session.flush()
    db.session.add(Computer(assetid=asset.assetid, hostname=hostname))
    # The enqueue route 404s without a reported state, so the endpoint has to
    # look like it has checked in at least once.
    from plugins.endpoints.models import EndpointState
    db.session.add(EndpointState(assetid=asset.assetid, hostname=hostname))
    db.session.commit()
    return asset.assetid


def _login(client, username, password='testpass'):
    response = client.post('/api/auth/login',
                           json={'username': username, 'password': password})
    token = response.get_json()['data']['access_token']
    return {'Authorization': f'Bearer {token}'}


def _user(db, client, username, permnames, admin=False):
    from shopdb.core.models import User, Role, Permission

    Permission.seed()
    Permission.seed_entries([
        ('endpoints.restart', 'Remote restart', 'endpoints'),
    ])
    db.session.flush()
    role = Role(rolename='admin' if admin else username + 'role')
    existing = Role.query.filter_by(rolename=role.rolename).first()
    if existing is not None:
        role = existing
    else:
        db.session.add(role)
    db.session.flush()
    role.permissions = Permission.query.filter(
        Permission.name.in_(permnames)).all()
    user = User(username=username, email=username + '@test.local',
                passwordhash=generate_password_hash('testpass'))
    user.roles.append(role)
    db.session.add(user)
    db.session.commit()
    return _login(client, username)


def _deny_everything(db, assetid):
    """A policy that matches this endpoint but permits only an unrelated action,
    so the action under test is denied by gating."""
    from plugins.computers.models import (ConfigurationPolicy,
                                          ConfigurationPolicyAction,
                                          BusinessPolicy, BusinessPolicyItem)
    policy = ConfigurationPolicy(name='Locked down', scopejson=None)
    db.session.add(policy)
    db.session.flush()
    db.session.add(ConfigurationPolicyAction(
        configurationpolicyid=policy.configurationpolicyid,
        action='mtconnect_status'))
    business = BusinessPolicy(name='Fleet')
    db.session.add(business)
    db.session.flush()
    db.session.add(BusinessPolicyItem(
        businesspolicyid=business.businesspolicyid,
        configurationpolicyid=policy.configurationpolicyid))
    db.session.commit()


def _url(assetid):
    return f'/api/endpoints/{assetid}/commands'


def test_policy_denial_still_applies_without_an_override(client, db):
    """The default path must be unchanged: no flag, no bypass."""
    _enable_policies(db)
    assetid = _endpoint(db)
    _deny_everything(db, assetid)
    headers = _user(db, client, 'plainadmin',
                    ['endpoints.view', 'endpoints.command'], admin=True)

    response = client.post(_url(assetid), headers=headers,
                           json={'action': 'service_status',
                                 'target': 'spooler'})
    assert response.status_code == 400, response.get_json()
    assert 'not permitted by policy' in \
        response.get_json()['data']['error']['message']


def test_a_non_admin_cannot_override_a_policy_denial(client, db):
    """endpoints.command is enough to queue work, never enough to break glass."""
    _enable_policies(db)
    assetid = _endpoint(db)
    _deny_everything(db, assetid)
    headers = _user(db, client, 'operator',
                    ['endpoints.view', 'endpoints.command'])

    response = client.post(_url(assetid), headers=headers,
                           json={'action': 'service_status',
                                 'target': 'spooler',
                                 'policyoverride': True,
                                 'overridereason': 'trying it on'})
    assert response.status_code == 403, response.get_json()
    assert 'admin' in response.get_json()['data']['error']['message']


def test_an_override_without_a_reason_is_refused(client, db):
    """An unexplained break-glass is an unreviewable one."""
    _enable_policies(db)
    assetid = _endpoint(db)
    _deny_everything(db, assetid)
    headers = _user(db, client, 'reasonless',
                    ['endpoints.view', 'endpoints.command'], admin=True)

    response = client.post(_url(assetid), headers=headers,
                           json={'action': 'service_status',
                                 'target': 'spooler',
                                 'policyoverride': True})
    assert response.status_code == 400, response.get_json()
    assert 'overridereason' in response.get_json()['data']['error']['message']


def test_an_admin_override_is_accepted_and_audited(client, db):
    """The whole point, plus the audit trail that justifies allowing it."""
    from shopdb.api import AuditLog

    _enable_policies(db)
    assetid = _endpoint(db)
    _deny_everything(db, assetid)
    headers = _user(db, client, 'breakglass',
                    ['endpoints.view', 'endpoints.command'], admin=True)

    response = client.post(_url(assetid), headers=headers,
                           json={'action': 'service_status',
                                 'target': 'spooler',
                                 'policyoverride': True,
                                 'overridereason': 'line down, need status'})
    assert response.status_code in (200, 201), response.get_json()

    entry = AuditLog.query.filter_by(action='policyoverridden').first()
    assert entry is not None, 'break-glass must leave an audit record'


def test_override_is_checked_after_the_elevated_permission_gate(client, db):
    """The override widens WHERE an action may run, never WHICH actions the
    caller may use.

    A human admin holds every permission via the admin bypass, so this is moot
    for people and cannot be shown with an admin fixture. It matters for TOKENS:
    authz suspends the admin bypass for a scoped personal token, so a token
    scoped to endpoints.command is refused at the elevated gate and never
    reaches the override branch at all - even though its owner is an admin, and
    even though it asks for the override explicitly.
    """
    from shopdb.core.models import Permission

    _enable_policies(db)
    assetid = _endpoint(db, 'tokenbox')
    _deny_everything(db, assetid)
    Permission.seed_entries([
        ('apitokens.create', 'Mint tokens', 'core'),
    ])
    db.session.commit()
    ownerheaders = _user(db, client, 'tokenadmin',
                         ['endpoints.view', 'endpoints.command',
                          'apitokens.create'], admin=True)

    minted = client.post('/api/apitokens', headers=ownerheaders,
                         json={'name': 'agent',
                               'scopes': ['endpoints.command']})
    secret = minted.get_json()['data']['secret']
    headers = {'Authorization': f'Bearer {secret}'}

    response = client.post(_url(assetid), headers=headers,
                           json={'action': 'system_restart',
                                 'policyoverride': True,
                                 'overridereason': 'leaked token tries it on'})
    assert response.status_code == 403, response.get_json()


def test_a_hold_suppresses_desired_state_reconciliation(client, db):
    """Without this the reconciler undoes the override on the next check-in."""
    from plugins.computers.models import PolicyActionHold
    from plugins.computers.services.policyresolver import effective_policy_for

    _enable_policies(db)
    assetid = _endpoint(db, 'desiredbox')

    from plugins.computers.models import (ConfigurationPolicy,
                                          ConfigurationPolicyAction,
                                          BusinessPolicy, BusinessPolicyItem)
    policy = ConfigurationPolicy(name='Keep it running', scopejson=None)
    db.session.add(policy)
    db.session.flush()
    db.session.add(ConfigurationPolicyAction(
        configurationpolicyid=policy.configurationpolicyid,
        action='service_start', mode='desired',
        desiredstate={'running': True}))
    business = BusinessPolicy(name='Desired fleet')
    db.session.add(business)
    db.session.flush()
    db.session.add(BusinessPolicyItem(
        businesspolicyid=business.businesspolicyid,
        configurationpolicyid=policy.configurationpolicyid))
    db.session.commit()

    assert 'service_start' in effective_policy_for(assetid).desired_actions()

    db.session.add(PolicyActionHold(
        assetid=assetid, action='service_start',
        expiresat=datetime.utcnow() + timedelta(minutes=60),
        reason='maintenance window'))
    db.session.commit()

    effective = effective_policy_for(assetid)
    assert 'service_start' not in effective.desired_actions()
    # A hold suspends drift correction; it does not revoke the permission to
    # run the action by hand.
    assert 'service_start' in effective.allowed_actions()


def test_an_expired_hold_stops_suppressing(client, db):
    """Policy resumes on its own. Nobody has to remember to clear a hold, and a
    stalled sweeper cannot silently disable reconciliation forever."""
    from plugins.computers.models import PolicyActionHold
    from plugins.computers.services.policyresolver import effective_policy_for

    _enable_policies(db)
    assetid = _endpoint(db, 'expiredbox')

    from plugins.computers.models import (ConfigurationPolicy,
                                          ConfigurationPolicyAction,
                                          BusinessPolicy, BusinessPolicyItem)
    policy = ConfigurationPolicy(name='Still running', scopejson=None)
    db.session.add(policy)
    db.session.flush()
    db.session.add(ConfigurationPolicyAction(
        configurationpolicyid=policy.configurationpolicyid,
        action='service_start', mode='desired',
        desiredstate={'running': True}))
    business = BusinessPolicy(name='Expiry fleet')
    db.session.add(business)
    db.session.flush()
    db.session.add(BusinessPolicyItem(
        businesspolicyid=business.businesspolicyid,
        configurationpolicyid=policy.configurationpolicyid))
    db.session.add(PolicyActionHold(
        assetid=assetid, action='service_start',
        expiresat=datetime.utcnow() - timedelta(minutes=1),
        reason='window that already closed'))
    db.session.commit()

    assert 'service_start' in effective_policy_for(assetid).desired_actions()


def test_active_holds_are_visible_to_a_non_admin_viewer(client, db):
    """A hold explains drift, so it must not be admin-only reading.

    An operator who can see the machine but not the hold is left debugging
    suppressed reconciliation as if it were a fault.
    """
    from plugins.computers.models import PolicyActionHold

    assetid = _endpoint(db)
    db.session.add(PolicyActionHold(
        assetid=assetid, action='service_stop',
        expiresat=datetime.utcnow() + timedelta(minutes=60),
        createdbyusername='nightadmin',
        reason='vendor on site'))
    db.session.commit()

    headers = _user(db, client, 'viewer', ['endpoints.view'])
    response = client.get(f'/api/endpoints/{assetid}/policyholds',
                          headers=headers)
    assert response.status_code == 200, response.get_json()
    holds = response.get_json()['data']
    assert len(holds) == 1
    assert holds[0]['action'] == 'service_stop'
    assert holds[0]['createdbyusername'] == 'nightadmin'
    assert holds[0]['reason'] == 'vendor on site'


def test_expired_holds_are_not_reported_as_active(client, db):
    """Expiry is evaluated on read, matching the resolver.

    If the route reported expired rows the banner would claim enforcement is
    paused long after it resumed - worse than showing nothing.
    """
    from plugins.computers.models import PolicyActionHold

    assetid = _endpoint(db)
    db.session.add(PolicyActionHold(
        assetid=assetid, action='service_stop',
        expiresat=datetime.utcnow() - timedelta(minutes=1),
        createdbyusername='nightadmin',
        reason='window that already closed'))
    db.session.commit()

    headers = _user(db, client, 'viewer2', ['endpoints.view'])
    response = client.get(f'/api/endpoints/{assetid}/policyholds',
                          headers=headers)
    assert response.status_code == 200, response.get_json()
    assert response.get_json()['data'] == []


def _desired_policy(db, assetid, action='service_start'):
    """A policy putting one action under desired-state enforcement."""
    from plugins.computers.models import (ConfigurationPolicy,
                                          ConfigurationPolicyAction,
                                          BusinessPolicy, BusinessPolicyItem)
    policy = ConfigurationPolicy(name=f'Desired {action}', scopejson=None)
    db.session.add(policy)
    db.session.flush()
    db.session.add(ConfigurationPolicyAction(
        configurationpolicyid=policy.configurationpolicyid,
        action=action, mode='desired', desiredstate={'running': True}))
    business = BusinessPolicy(name=f'Fleet {action}')
    db.session.add(business)
    db.session.flush()
    db.session.add(BusinessPolicyItem(
        businesspolicyid=business.businesspolicyid,
        configurationpolicyid=policy.configurationpolicyid))
    db.session.commit()


def test_an_admin_can_place_a_hold_directly(client, db):
    """The hold must be reachable deliberately, not only by tripping a denial."""
    _enable_policies(db)
    assetid = _endpoint(db)
    _desired_policy(db, assetid)
    headers = _user(db, client, 'holdadmin',
                    ['endpoints.view', 'endpoints.command'], admin=True)

    response = client.post(f'/api/endpoints/{assetid}/policyholds',
                           headers=headers,
                           json={'action': 'service_start',
                                 'reason': 'vendor on site',
                                 'minutes': 60})
    assert response.status_code == 200, response.get_json()
    holds = response.get_json()['data']
    assert [h['action'] for h in holds] == ['service_start']

    # The point of a hold: reconciliation stops asking for this action.
    from plugins.computers.services.policyresolver import effective_policy_for
    assert 'service_start' not in effective_policy_for(assetid).desired_actions()


def test_a_non_admin_cannot_place_a_hold(client, db):
    """endpoints.command queues work; it does not switch off enforcement."""
    _enable_policies(db)
    assetid = _endpoint(db)
    _desired_policy(db, assetid)
    headers = _user(db, client, 'holdoperator',
                    ['endpoints.view', 'endpoints.command'])

    response = client.post(f'/api/endpoints/{assetid}/policyholds',
                           headers=headers,
                           json={'action': 'service_start',
                                 'reason': 'trying it on'})
    assert response.status_code == 403, response.get_json()


def test_a_hold_requires_a_reason(client, db):
    """An unexplained hold is an unattributable outage waiting to happen."""
    _enable_policies(db)
    assetid = _endpoint(db)
    _desired_policy(db, assetid)
    headers = _user(db, client, 'holdadmin2',
                    ['endpoints.view', 'endpoints.command'], admin=True)

    response = client.post(f'/api/endpoints/{assetid}/policyholds',
                           headers=headers,
                           json={'action': 'service_start', 'reason': '   '})
    assert response.status_code == 400, response.get_json()


def test_holding_a_non_desired_action_is_refused(client, db):
    """A hold on a gate-mode action would suppress nothing.

    Accepting it would report a hold that has no effect, which is worse than
    refusing outright.
    """
    _enable_policies(db)
    assetid = _endpoint(db)
    _deny_everything(db, assetid)
    headers = _user(db, client, 'holdadmin3',
                    ['endpoints.view', 'endpoints.command'], admin=True)

    response = client.post(f'/api/endpoints/{assetid}/policyholds',
                           headers=headers,
                           json={'action': 'mtconnect_status',
                                 'reason': 'not under enforcement'})
    assert response.status_code == 400, response.get_json()
    assert 'nothing to hold' in \
        response.get_json()['data']['error']['message']


def test_an_admin_can_release_a_hold_early(client, db):
    """Release resumes reconciliation without waiting for expiry."""
    _enable_policies(db)
    assetid = _endpoint(db)
    _desired_policy(db, assetid)
    headers = _user(db, client, 'holdadmin4',
                    ['endpoints.view', 'endpoints.command'], admin=True)

    client.post(f'/api/endpoints/{assetid}/policyholds', headers=headers,
                json={'action': 'service_start', 'reason': 'swapping a disk'})

    from plugins.computers.services.policyresolver import effective_policy_for
    assert 'service_start' not in effective_policy_for(assetid).desired_actions()

    response = client.delete(
        f'/api/endpoints/{assetid}/policyholds/service_start', headers=headers)
    assert response.status_code == 200, response.get_json()
    assert response.get_json()['data'] == []
    assert 'service_start' in effective_policy_for(assetid).desired_actions()


def test_releasing_a_hold_keeps_the_audit_row(client, db):
    """Release soft-deletes: the hold IS the record of what was suppressed."""
    from plugins.computers.models import PolicyActionHold

    _enable_policies(db)
    assetid = _endpoint(db)
    _desired_policy(db, assetid)
    headers = _user(db, client, 'holdadmin5',
                    ['endpoints.view', 'endpoints.command'], admin=True)

    client.post(f'/api/endpoints/{assetid}/policyholds', headers=headers,
                json={'action': 'service_start', 'reason': 'keep this record'})
    client.delete(f'/api/endpoints/{assetid}/policyholds/service_start',
                  headers=headers)

    row = PolicyActionHold.query.filter_by(assetid=assetid,
                                           action='service_start').first()
    assert row is not None
    assert row.isactive is False
    assert row.reason == 'keep this record'


def test_holdable_actions_lists_desired_actions_with_hold_state(client, db):
    """The picker must keep showing an action after it is held.

    Built from desired_actions() it would vanish on being held, making the
    current state unreadable.
    """
    _enable_policies(db)
    assetid = _endpoint(db)
    _desired_policy(db, assetid)
    headers = _user(db, client, 'holdadmin6',
                    ['endpoints.view', 'endpoints.command'], admin=True)

    response = client.get(f'/api/endpoints/{assetid}/holdableactions',
                          headers=headers)
    assert response.status_code == 200, response.get_json()
    assert response.get_json()['data'] == [
        {'action': 'service_start', 'held': False}]

    client.post(f'/api/endpoints/{assetid}/policyholds', headers=headers,
                json={'action': 'service_start', 'reason': 'still listed'})

    response = client.get(f'/api/endpoints/{assetid}/holdableactions',
                          headers=headers)
    assert response.get_json()['data'] == [
        {'action': 'service_start', 'held': True}]


def _lock_url(assetid):
    return f'/api/endpoints/{assetid}/lock'


def test_a_lock_refuses_a_non_admin_command(client, db):
    """The whole point: a locked endpoint stops other people acting on it.

    No policy is enabled here, so nothing but the lock can be doing the
    refusing - otherwise this test would pass for the wrong reason.
    """
    assetid = _endpoint(db)
    adminheaders = _user(db, client, 'lockadmin',
                         ['endpoints.view', 'endpoints.command'], admin=True)
    client.post(_lock_url(assetid), headers=adminheaders,
                json={'reason': 'mid repair'})

    headers = _user(db, client, 'lockoperator',
                    ['endpoints.view', 'endpoints.command'])
    response = client.post(_url(assetid), headers=headers,
                           json={'action': 'service_status',
                                 'target': 'spooler'})
    assert response.status_code == 403, response.get_json()


def test_a_lock_refusal_names_the_owner_and_reason(client, db):
    """A bare "forbidden" sends an operator hunting through policy for a cause
    that is not there. The refusal has to say who to ask and why."""
    assetid = _endpoint(db)
    adminheaders = _user(db, client, 'lockadmin2',
                         ['endpoints.view', 'endpoints.command'], admin=True)
    client.post(_lock_url(assetid), headers=adminheaders,
                json={'reason': 'validated part running'})

    headers = _user(db, client, 'lockoperator2',
                    ['endpoints.view', 'endpoints.command'])
    response = client.post(_url(assetid), headers=headers,
                           json={'action': 'service_status',
                                 'target': 'spooler'})
    message = response.get_json()['data']['error']['message']
    assert 'lockadmin2' in message
    assert 'validated part running' in message


def test_an_admin_acts_through_a_lock_without_an_override(client, db):
    """The lock gates WHO may act, not WHICH actions exist, so there is no
    policy decision for an admin to break past. Requiring break-glass here would
    mean an admin overriding their own quarantine to do the repair it was set
    for."""
    assetid = _endpoint(db)
    headers = _user(db, client, 'lockadmin3',
                    ['endpoints.view', 'endpoints.command'], admin=True)
    client.post(_lock_url(assetid), headers=headers,
                json={'reason': 'mine to work on'})

    response = client.post(_url(assetid), headers=headers,
                           json={'action': 'service_status',
                                 'target': 'spooler'})
    assert response.status_code == 201, response.get_json()


def test_a_lock_cannot_be_broken_by_a_policy_override(client, db):
    """The two gates are independent and the lock is the outer one.

    policyoverride exists to push past a POLICY denial. If it also dissolved a
    lock, any non-admin who learned the flag could act on a quarantined machine,
    and the lock would be advisory.
    """
    _enable_policies(db)
    assetid = _endpoint(db)
    _deny_everything(db, assetid)
    adminheaders = _user(db, client, 'lockadmin4',
                         ['endpoints.view', 'endpoints.command'], admin=True)
    client.post(_lock_url(assetid), headers=adminheaders,
                json={'reason': 'do not touch'})

    headers = _user(db, client, 'lockoperator4',
                    ['endpoints.view', 'endpoints.command'])
    response = client.post(_url(assetid), headers=headers,
                           json={'action': 'service_status',
                                 'target': 'spooler',
                                 'policyoverride': True,
                                 'overridereason': 'trying to slip past'})
    assert response.status_code == 403, response.get_json()
    assert 'locked' in response.get_json()['data']['error']['message'].lower()


def test_a_non_admin_cannot_place_a_lock(client, db):
    assetid = _endpoint(db)
    headers = _user(db, client, 'lockoperator5',
                    ['endpoints.view', 'endpoints.command'])

    response = client.post(_lock_url(assetid), headers=headers,
                           json={'reason': 'not my call'})
    assert response.status_code == 403, response.get_json()


def test_a_lock_requires_a_reason(client, db):
    """The reason is the message a blocked operator reads, not paperwork."""
    assetid = _endpoint(db)
    headers = _user(db, client, 'lockadmin6',
                    ['endpoints.view', 'endpoints.command'], admin=True)

    response = client.post(_lock_url(assetid), headers=headers,
                           json={'reason': '   '})
    assert response.status_code == 400, response.get_json()


def test_locking_twice_does_not_stack_a_second_lock(client, db):
    """Two active locks on one endpoint is not a state the table should hold,
    and the caller who locked a locked machine got what they asked for."""
    assetid = _endpoint(db)
    headers = _user(db, client, 'lockadmin7',
                    ['endpoints.view', 'endpoints.command'], admin=True)

    first = client.post(_lock_url(assetid), headers=headers,
                        json={'reason': 'first'})
    second = client.post(_lock_url(assetid), headers=headers,
                         json={'reason': 'second'})
    assert second.status_code == 200, second.get_json()
    # The original stands - the second call did not silently replace the reason
    # the first admin recorded.
    assert second.get_json()['data']['endpointlockid'] == \
        first.get_json()['data']['endpointlockid']
    assert second.get_json()['data']['reason'] == 'first'


def test_releasing_a_lock_restores_non_admin_commands(client, db):
    assetid = _endpoint(db)
    adminheaders = _user(db, client, 'lockadmin8',
                         ['endpoints.view', 'endpoints.command'], admin=True)
    client.post(_lock_url(assetid), headers=adminheaders,
                json={'reason': 'briefly'})
    client.delete(_lock_url(assetid), headers=adminheaders)

    headers = _user(db, client, 'lockoperator8',
                    ['endpoints.view', 'endpoints.command'])
    response = client.post(_url(assetid), headers=headers,
                           json={'action': 'service_status',
                                 'target': 'spooler'})
    assert response.status_code == 201, response.get_json()


def test_releasing_a_lock_keeps_the_row_as_history(client, db):
    """Released by stamping releasedat, not by deleting. The record of who
    quarantined a machine is most wanted after the lock is gone and no longer
    there to explain itself."""
    from plugins.endpoints.models import EndpointLock

    assetid = _endpoint(db)
    headers = _user(db, client, 'lockadmin9',
                    ['endpoints.view', 'endpoints.command'], admin=True)
    client.post(_lock_url(assetid), headers=headers,
                json={'reason': 'keep this record'})
    client.delete(_lock_url(assetid), headers=headers)

    row = EndpointLock.query.filter_by(assetid=assetid).first()
    assert row is not None
    assert row.reason == 'keep this record'
    assert row.lockedbyusername == 'lockadmin9'
    assert row.releasedat is not None
    assert row.releasedbyusername == 'lockadmin9'


def test_releasing_an_unlocked_endpoint_is_a_404(client, db):
    """So a caller can tell "unlocked it" from "there was nothing to unlock"."""
    assetid = _endpoint(db)
    headers = _user(db, client, 'lockadmin10',
                    ['endpoints.view', 'endpoints.command'], admin=True)

    response = client.delete(_lock_url(assetid), headers=headers)
    assert response.status_code == 404, response.get_json()


def test_a_non_admin_can_read_the_lock(client, db):
    """Read-gated, not admin-gated. An operator who cannot see the lock is left
    with a bare refusal and files a ticket about a machine that is working."""
    assetid = _endpoint(db)
    adminheaders = _user(db, client, 'lockadmin11',
                         ['endpoints.view', 'endpoints.command'], admin=True)
    client.post(_lock_url(assetid), headers=adminheaders,
                json={'reason': 'visible to all'})

    headers = _user(db, client, 'lockoperator11', ['endpoints.view'])
    response = client.get(_lock_url(assetid), headers=headers)
    assert response.status_code == 200, response.get_json()
    assert response.get_json()['data']['reason'] == 'visible to all'


def test_an_unlocked_endpoint_reports_no_lock(client, db):
    assetid = _endpoint(db)
    headers = _user(db, client, 'lockoperator12', ['endpoints.view'])

    response = client.get(_lock_url(assetid), headers=headers)
    assert response.status_code == 200, response.get_json()
    # success_response(None) omits data entirely rather than emitting null, so
    # "no lock" is an absent key. The frontend reads it as response.data?.data
    # || null, which lands on the same answer either way.
    assert response.get_json().get('data') is None


def test_a_lock_does_not_suppress_policy_reconciliation(client, db):
    """The two axes stay separate. A lock gates callers; it must not reach into
    the resolver and start suppressing desired-state actions, which is the hold's
    job and carries an expiry a lock does not have."""
    from plugins.computers.services.policyresolver import effective_policy_for

    _enable_policies(db)
    assetid = _endpoint(db)
    _desired_policy(db, assetid)
    headers = _user(db, client, 'lockadmin13',
                    ['endpoints.view', 'endpoints.command'], admin=True)
    client.post(_lock_url(assetid), headers=headers,
                json={'reason': 'locked but still reconciled'})

    assert 'service_start' in effective_policy_for(assetid).desired_actions()


def test_an_admin_can_hold_several_actions_at_once(client, db):
    """One dialog submission covers the whole job, not one action per trip.

    An admin about to work on a machine usually needs several actions pinned,
    and making them repeat the dialog invites stopping halfway.
    """
    from plugins.computers.services.policyresolver import effective_policy_for

    _enable_policies(db)
    assetid = _endpoint(db)
    _desired_policy(db, assetid, action='service_start')
    _desired_policy(db, assetid, action='service_stop')
    headers = _user(db, client, 'multiadmin1',
                    ['endpoints.view', 'endpoints.command'], admin=True)

    response = client.post(f'/api/endpoints/{assetid}/policyholds',
                           headers=headers,
                           json={'actions': ['service_start', 'service_stop'],
                                 'reason': 'stripping the enclosure',
                                 'minutes': 60})
    assert response.status_code == 200, response.get_json()
    held = {h['action'] for h in response.get_json()['data']}
    assert held == {'service_start', 'service_stop'}

    desired = effective_policy_for(assetid).desired_actions()
    assert 'service_start' not in desired
    assert 'service_stop' not in desired


def test_a_multi_action_hold_is_all_or_nothing(client, db):
    """One bad name rejects the request whole.

    Partial success is the wrong failure mode on a break-glass control: the
    admin walks away believing the machine is pinned, and the reconciler undoes
    the work covered by the holds that never landed.
    """
    from plugins.computers.models import PolicyActionHold

    _enable_policies(db)
    assetid = _endpoint(db)
    _desired_policy(db, assetid, action='service_start')
    headers = _user(db, client, 'multiadmin2',
                    ['endpoints.view', 'endpoints.command'], admin=True)

    response = client.post(
        f'/api/endpoints/{assetid}/policyholds',
        headers=headers,
        # service_start is holdable; mtconnect_status is not under enforcement.
        json={'actions': ['service_start', 'mtconnect_status'],
              'reason': 'half of this is invalid'})
    assert response.status_code == 400, response.get_json()

    # The valid half must NOT have been written on the way to the refusal.
    assert PolicyActionHold.query.filter_by(assetid=assetid).count() == 0


def test_a_zero_minute_hold_never_expires(client, db):
    """0 means indefinite: it outlives any window a dated hold would have."""
    from plugins.computers.models import PolicyActionHold
    from plugins.computers.services.policyresolver import effective_policy_for

    _enable_policies(db)
    assetid = _endpoint(db)
    _desired_policy(db, assetid)
    headers = _user(db, client, 'indefadmin1',
                    ['endpoints.view', 'endpoints.command'], admin=True)

    response = client.post(f'/api/endpoints/{assetid}/policyholds',
                           headers=headers,
                           json={'actions': ['service_start'],
                                 'reason': 'awaiting a replacement part',
                                 'minutes': 0})
    assert response.status_code == 200, response.get_json()

    row = PolicyActionHold.query.filter_by(assetid=assetid,
                                           action='service_start').first()
    assert row.expiresat is None
    # Reported active and honoured by the resolver despite having no expiry.
    assert response.get_json()['data'][0]['expiresat'] is None
    assert 'service_start' not in effective_policy_for(assetid).desired_actions()


def test_an_indefinite_hold_can_still_be_released(client, db):
    """The one hold that never self-clears must be clearable by hand.

    Without a null branch in the release query this would be the only kind of
    hold the release button could not touch - permanently stuck.
    """
    from plugins.computers.services.policyresolver import effective_policy_for

    _enable_policies(db)
    assetid = _endpoint(db)
    _desired_policy(db, assetid)
    headers = _user(db, client, 'indefadmin2',
                    ['endpoints.view', 'endpoints.command'], admin=True)
    client.post(f'/api/endpoints/{assetid}/policyholds', headers=headers,
                json={'actions': ['service_start'], 'reason': 'indefinite',
                      'minutes': 0})

    response = client.delete(
        f'/api/endpoints/{assetid}/policyholds/service_start', headers=headers)
    assert response.status_code == 200, response.get_json()
    assert 'service_start' in effective_policy_for(assetid).desired_actions()


def test_a_negative_hold_window_is_refused(client, db):
    """A negative window places an already-expired hold.

    That reads as success in the UI while suppressing nothing - the silent
    failure the holdable-action check exists to prevent.
    """
    _enable_policies(db)
    assetid = _endpoint(db)
    _desired_policy(db, assetid)
    headers = _user(db, client, 'indefadmin3',
                    ['endpoints.view', 'endpoints.command'], admin=True)

    response = client.post(f'/api/endpoints/{assetid}/policyholds',
                           headers=headers,
                           json={'actions': ['service_start'],
                                 'reason': 'backwards', 'minutes': -30})
    assert response.status_code == 400, response.get_json()


def test_re_holding_an_active_action_does_not_stack_rows(client, db):
    """Idempotent, matching the endpoint lock.

    A duplicate row would have to be released once per hidden copy before
    reconciliation actually resumed.
    """
    from plugins.computers.models import PolicyActionHold
    from plugins.computers.services.policyresolver import effective_policy_for

    _enable_policies(db)
    assetid = _endpoint(db)
    _desired_policy(db, assetid)
    headers = _user(db, client, 'dupeadmin1',
                    ['endpoints.view', 'endpoints.command'], admin=True)

    for _ in range(2):
        client.post(f'/api/endpoints/{assetid}/policyholds', headers=headers,
                    json={'actions': ['service_start'], 'reason': 'twice'})

    assert PolicyActionHold.query.filter_by(assetid=assetid,
                                            action='service_start').count() == 1

    # One release is therefore enough to resume reconciliation.
    client.delete(f'/api/endpoints/{assetid}/policyholds/service_start',
                  headers=headers)
    assert 'service_start' in effective_policy_for(assetid).desired_actions()


def test_a_single_action_hold_payload_still_works(client, db):
    """The pre-list payload shape stays accepted so older callers keep working."""
    _enable_policies(db)
    assetid = _endpoint(db)
    _desired_policy(db, assetid)
    headers = _user(db, client, 'legacyadmin1',
                    ['endpoints.view', 'endpoints.command'], admin=True)

    response = client.post(f'/api/endpoints/{assetid}/policyholds',
                           headers=headers,
                           json={'action': 'service_start',
                                 'reason': 'legacy payload'})
    assert response.status_code == 200, response.get_json()
    assert [h['action'] for h in response.get_json()['data']] == \
        ['service_start']