diff --git a/unit_tests/test_zaza_model.py b/unit_tests/test_zaza_model.py index c3d4e3466..b7b5b512f 100644 --- a/unit_tests/test_zaza_model.py +++ b/unit_tests/test_zaza_model.py @@ -1023,6 +1023,15 @@ async def _get_status(): 'active', timeout=0.1) + def test_get_file_stats(self): + self.action.data = { + 'results': {'Code': '0', 'Stderr': '', 'Stdout': '444'} + } + self.patch_object(model, 'Model') + self.Model.return_value = self.Model_mock + mtime = model.get_file_stats('%Y', 'app/2', '/etc/myfile.cnf') + self.assertEqual(mtime, '444') + def test_wait_for_agent_status(self): async def _block_until(f, timeout=None): if not f(): diff --git a/zaza/model.py b/zaza/model.py index 04900d7c3..0f78ef708 100644 --- a/zaza/model.py +++ b/zaza/model.py @@ -21,6 +21,7 @@ import asyncio from async_generator import async_generator, yield_, asynccontextmanager +import functools import logging import os import subprocess @@ -684,6 +685,34 @@ def check_unit_workload_status_message(model, unit, message=None, raise ValueError("Must be called with message or prefixes") +async def aync_get_file_stats(stat_format, unit_name, remote_file, + model_name=None, timeout=2700): + """Stat the given file on a unit and return the results. + + :param stat_format: Format for stat command. + :type stat_format: str + :param unit_name: Name of unit to run action on + :type unit_name: str + :param remote_file: Remote file to stat. + :type remote_file: str + :param model_name: Name of model to query. + :type model_name: str + :param timeout: Time to wait for command to run on remote unit. + :type timeout: int + """ + async with run_in_model(model_name) as model: + unit = get_unit_from_name(unit_name, model=model) + output = await unit.run( + 'stat --format "{}" {}'.format(stat_format, remote_file)) + return output.data.get('results')['Stdout'].strip() + +get_file_stats = sync_wrapper(aync_get_file_stats) + +get_file_mtime = functools.partial(get_file_stats, '%Y') + +get_file_ctime = functools.partial(get_file_stats, '%W') + + async def async_wait_for_agent_status(model_name=None, status='executing', timeout=60): """Wait for at least one unit to enter a specific agent status.