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_extract_sql_context():
text = "\n".join([f"{n}" for n in range(1, 21)])
expected_result = "| 4\n| 5\n* 6\n| 7\n| 8"
result = printer.extract_sql_context(sql=text, line_number=6, window_size=2)
assert delete_color_codes(result) == expected_result
def test_mark_line_even_number_of_lines():
text = [f"{n}" for n in range(1, 5)]
expected_result = ["| 1", "* 2", "| 3", "| 4"]
result = printer.mark_line(lines=text, line_number=2)
result = [delete_color_codes(line) for line in result]
assert result == expected_result
def wrapper(*args, **kwargs):
try:
return function(*args, **kwargs)
except GenericValidationError as error:
sys.exit(error.exit_code)
except LookerApiError as error:
logger.error(
f"\n{error}\n\n"
+ printer.dim(
"Run in verbose mode (-v) or check your log file to see the full "
"response from the Looker API. "
"For support, please create an issue at "
"https://github.com/spectacles-ci/spectacles/issues"
)
+ "\n"
)
looker_api_response = json.dumps(error.looker_api_response, indent=2)
logger.debug(
f"Spectacles received a {error.status} response code from "
f"the Looker API with the following details: {looker_api_response}\n"
)
sys.exit(error.exit_code)
except SpectaclesException as error:
logger.error(
f"\n{error}\n\n"
import_projects,
commit_ref,
)
results = runner.validate_content(explores, excludes, incremental, exclude_personal)
for test in sorted(results["tested"], key=lambda x: (x["model"], x["explore"])):
message = f"{test['model']}.{test['explore']}"
printer.print_validation_result(passed=test["passed"], source=message)
errors = sorted(
results["errors"],
key=lambda x: (x["model"], x["explore"], x["metadata"]["field_name"]),
)
if errors:
for error in errors:
printer.print_content_error(
error["model"],
error["explore"],
error["message"],
error["metadata"]["content_type"],
error["metadata"]["space"],
error["metadata"]["title"],
error["metadata"]["url"],
)
logger.info("")
raise GenericValidationError
else:
logger.info("")
)
if errors:
for error in errors:
printer.print_sql_error(
model=error["model"],
explore=error["explore"],
message=error["message"],
sql=error["metadata"]["sql"],
log_dir=log_dir,
dimension=error["metadata"].get("dimension"),
lookml_url=error["metadata"].get("lookml_url"),
)
if mode == "batch":
logger.info(
printer.dim(
"\n\nTo determine the exact dimensions responsible for "
f"{'this error' if len(errors) == 1 else 'these errors'}, "
"you can re-run \nSpectacles in single-dimension mode, "
"with `--mode single`.\n\nYou can also run this original "
"validation with `--mode hybrid` to do this automatically."
)
)
logger.info("")
raise GenericValidationError
else:
logger.info("")
import_projects,
commit_ref,
)
results = runner.validate_data_tests(explores, exclude)
for test in sorted(results["tested"], key=lambda x: (x["model"], x["explore"])):
message = f"{test['model']}.{test['explore']}"
printer.print_validation_result(passed=test["passed"], source=message)
errors = sorted(
results["errors"],
key=lambda x: (x["model"], x["explore"], x["metadata"]["test_name"]),
)
if errors:
for error in errors:
printer.print_data_test_error(
error["model"],
error["explore"],
error["metadata"]["test_name"],
error["message"],
error["metadata"]["lookml_url"],
)
logger.info("")
raise GenericValidationError
else:
logger.info("")