Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def deploy_models(self, username: str, password: str):
repo_dir = os.path.abspath(os.path.dirname(tabpy.__file__))
path = os.path.join(repo_dir, "models", "deploy_models.py")
with open(self.tmp_dir + "/deploy_models_output.txt", "w") as outfile:
outfile.write(
f"--<< Running {self.py} {path} "
f"{self._get_config_file_name()} >>--\n"
)
input_string = f"{username}\n{password}\n"
outfile.write(f"--<< Input = {input_string} >>--")
coverage.process_startup()
p = subprocess.run(
[self.py, path, self._get_config_file_name()],
input=input_string.encode("utf-8"),
stdout=outfile,
stderr=outfile,
)
def test_no_config_file(
self,
mock_os,
mock_path_exists,
mock_psws,
mock_management_util,
mock_tabpy_state,
mock_parse_arguments,
):
pkg_path = os.path.dirname(tabpy.__file__)
obj_path = os.path.join(pkg_path, "tmp", "query_objects")
state_path = os.path.join(pkg_path, "tabpy_server")
mock_os.environ = {
"TABPY_PORT": "9004",
"TABPY_QUERY_OBJECT_PATH": obj_path,
"TABPY_STATE_PATH": state_path,
}
TabPyApp(None)
self.assertEqual(len(mock_psws.mock_calls), 1)
self.assertEqual(len(mock_tabpy_state.mock_calls), 1)
self.assertEqual(len(mock_path_exists.mock_calls), 1)
self.assertTrue(len(mock_management_util.mock_calls) > 0)
mock_os.makedirs.assert_not_called()
def read_version():
ver = "unknown"
import tabpy
pkg_path = os.path.dirname(tabpy.__file__)
ver_file_path = os.path.join(pkg_path, "VERSION")
if Path(ver_file_path).exists():
with open(ver_file_path) as f:
ver = f.read().strip()
else:
ver = f"Version Unknown, (file {ver_file_path} not found)"
return ver
def get_default_config_file_path():
import tabpy
pkg_path = os.path.dirname(tabpy.__file__)
config_file_path = os.path.join(pkg_path, "tabpy_server", "common", "default.conf")
return config_file_path
ConfigParameters.TABPY_EVALUATE_TIMEOUT,
default_val=30,
)
try:
self.settings[SettingsParameters.EvaluateTimeout] = float(
self.settings[SettingsParameters.EvaluateTimeout]
)
except ValueError:
logger.warning(
"Evaluate timeout must be a float type. Defaulting "
"to evaluate timeout of 30 seconds."
)
self.settings[SettingsParameters.EvaluateTimeout] = 30
pkg_path = os.path.dirname(tabpy.__file__)
set_parameter(
SettingsParameters.UploadDir,
ConfigParameters.TABPY_QUERY_OBJECT_PATH,
default_val=os.path.join(pkg_path, "tmp", "query_objects"),
)
if not os.path.exists(self.settings[SettingsParameters.UploadDir]):
os.makedirs(self.settings[SettingsParameters.UploadDir])
# set and validate transfer protocol
set_parameter(
SettingsParameters.TransferProtocol,
ConfigParameters.TABPY_TRANSFER_PROTOCOL,
default_val="http",
)
self.settings[SettingsParameters.TransferProtocol] = self.settings[
SettingsParameters.TransferProtocol