import time from frostfs_testlib import reporter from frostfs_testlib.resources.common import STORAGE_GC_TIME from frostfs_testlib.utils import datetime_utils def placement_policy_from_container(container_info: str) -> str: """ Get placement policy from container info: container ID: j7k4auNHRmiPMSmnH2qENLECD2au2y675fvTX6csDwd version: 2.12 owner ID: NQ8HUxE5qEj7UUvADj7z9Z7pcvJdjtPwuw basic ACL: 0fbfbfff (eacl-public-read-write) attribute: Timestamp=1656340345 (2022-06-27 17:32:25 +0300 MSK) nonce: 1c511e88-efd7-4004-8dbf-14391a5d375a placement policy: REP 1 IN LOC_PLACE CBF 1 SELECT 1 FROM LOC_SW AS LOC_PLACE FILTER Country EQ Sweden AS LOC_SW Args: container_info: output from frostfs-cli container get command Returns: placement policy as a string """ assert ":" in container_info, f"Could not find placement rule in the output {container_info}" return container_info.split(":")[-1].replace("\n", " ").strip() def wait_for_gc_pass_on_storage_nodes() -> None: wait_time = datetime_utils.parse_time(STORAGE_GC_TIME) with reporter.step(f"Wait {wait_time}s until GC completes on storage nodes"): time.sleep(wait_time)