Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_full_redis_check_error(mocker):
app = Flask("redis-check")
redis_store = FlaskRedis.from_custom_provider(FakeStrictRedis, app)
ping = mocker.patch.object(redis_store, "ping")
ping.side_effect = redis.ConnectionError
dockerflow = Dockerflow(app, redis=redis_store)
assert "check_redis_connected" in dockerflow.checks
with app.test_client() as test_client:
response = test_client.get("/__heartbeat__")
assert response.status_code == 500
assert json.loads(response.data.decode())["status"] == "error"
def dockerflow(app):
return Dockerflow(app)
def test_version_path(mocker, version_content):
app = Flask("dockerflow")
app.secret_key = "super sekrit"
login_manager = LoginManager(app)
login_manager.user_loader(load_user)
custom_version_path = "/something/extra/ordinary"
dockerflow = Dockerflow(app, version_path=custom_version_path)
version_callback = mocker.patch.object(
dockerflow, "_version_callback", return_value=version_content
)
with app.test_client() as test_client:
response = test_client.get("/__version__")
assert response.status_code == 200
assert json.loads(response.data.decode()) == version_content
version_callback.assert_called_with(custom_version_path)
def test_instantiating(app):
dockerflow = Dockerflow()
assert "dockerflow.heartbeat" not in app.view_functions
dockerflow.init_app(app)
assert "dockerflow.heartbeat" in app.view_functions
def test_request_summary_exception(caplog, app):
Dockerflow(app)
with app.test_request_context("/", headers=headers):
assert has_request_context()
app.preprocess_request()
app.handle_exception(ValueError("exception message"))
response = Response("")
response = app.process_response(response)
for record in caplog.records:
if record != "request.summary":
continue
assert_log_record(request, record, level=logging.ERROR, errno=500)
assert record.getMessage() == "exception message"
def test_full_db_check_error(mocker):
app = Flask("db-check")
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite://"
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
db = SQLAlchemy(app)
engine_connect = mocker.patch.object(db.engine, "connect")
engine_connect.side_effect = SQLAlchemyError
dockerflow = Dockerflow(app, db=db)
assert "check_database_connected" in dockerflow.checks
with app.test_client() as test_client:
response = test_client.get("/__heartbeat__")
assert response.status_code == 500
assert json.loads(response.data.decode())["status"] == "error"
def test_full_db_check(mocker):
app = Flask("db-check")
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite://"
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
db = SQLAlchemy(app)
dockerflow = Dockerflow(app, db=db)
assert "check_database_connected" in dockerflow.checks
response = app.test_client().get("/__heartbeat__")
assert response.status_code == 200
assert json.loads(response.data.decode())["status"] == "ok"
def test_full_redis_check(mocker):
app = Flask("redis-check")
app.debug = True
redis_store = FlaskRedis.from_custom_provider(FakeStrictRedis, app)
dockerflow = Dockerflow(app, redis=redis_store)
assert "check_redis_connected" in dockerflow.checks
with app.test_client() as test_client:
response = test_client.get("/__heartbeat__")
assert response.status_code == 200
assert json.loads(response.data.decode())["status"] == "ok"