"""Mint a temporary collector token for the load test, or revoke it after. WHY a minted token: plugin.endpoints.envkeyfallback is false, which is the agent kill switch working as intended - the env key no longer authorizes. The harness therefore needs a real managed token scoped collector.ingest.endpoints, exactly like a live agent. The token is named LOADTEST so revoke() removes only this one and never touches the real agent keys. Usage (inside the api container, with PYTHONPATH=/app and cwd /app): python loadtest_token.py mint # prints the secret once python loadtest_token.py revoke """ import sys from shopdb import create_app, db from shopdb.core.models.apitoken import ApiToken TOKEN_NAME = 'LOADTEST temporary collector token' SCOPE = 'collector.ingest.endpoints' def mint(): existing = ApiToken.query.filter_by(name=TOKEN_NAME).first() if existing is not None: db.session.delete(existing) db.session.commit() owner = ApiToken.query.filter( ApiToken.scopes.like('%' + SCOPE + '%')).first() if owner is None: raise SystemExit('no existing collector token to copy ownership from') secret = ApiToken.generate_secret() token = ApiToken( userid=owner.userid, name=TOKEN_NAME, tokenprefix=ApiToken.prefix_of(secret), tokenhash=ApiToken.hash_secret(secret), ) token.scopelist = [SCOPE] db.session.add(token) db.session.commit() print(secret) def revoke(): removed = ApiToken.query.filter_by(name=TOKEN_NAME).delete() db.session.commit() print('revoked %d loadtest token(s)' % removed) if __name__ == '__main__': action = sys.argv[1] if len(sys.argv) > 1 else '' app = create_app() with app.app_context(): if action == 'mint': mint() elif action == 'revoke': revoke() else: print(__doc__) sys.exit(2)