Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use release_container method in autotest

Best Python code snippet using autotest_python

mag.py

Source:mag.py

1# Copyright 2020 Curtin University2#3# Licensed under the Apache License, Version 2.0 (the "License");4# you may not use this file except in compliance with the License.5# You may obtain a copy of the License at6#7#   http://www.apache.org/licenses/LICENSE-2.08#9# Unless required by applicable law or agreed to in writing, software10# distributed under the License is distributed on an "AS IS" BASIS,11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12# See the License for the specific language governing permissions and13# limitations under the License.14# Author: James Diprose15import logging16import re17from enum import Enum18from typing import Any, List, Optional19import pendulum20from azure.common import AzureConflictHttpError21from azure.cosmosdb.table.models import EdmType, EntityProperty22from azure.cosmosdb.table.tableservice import TableService23from azure.storage.blob import ContainerProperties24from mag_archiver.azure import copy_container, delete_container, list_containers25class MagState(Enum):26    """The state of the MAG release"""27    discovered = "discovered"28    archived = "archived"29    done = "done"30class MagTask(Enum):31    """The current task being executed"""32    not_started = "not-started"33    copying_to_vm = "copying-to-vm"34    archiving = "archiving"35    copying_to_release_container = "copying-to-release-container"36    cleaning_up = "cleaning-up"37    done = "done"38class MagDateType(Enum):39    """The date field type"""40    release = "ReleaseDate"41    discovered = "DiscoveredDate"42    archived = "ArchivedDate"43    done = "DoneDate"44    @classmethod45    def map(cls):46        return {47            MagDateType.release: "release_date",48            MagDateType.discovered: "discovered_date",49            MagDateType.archived: "archived_date",50            MagDateType.done: "done_date",51        }52    def attr(self):53        map_ = MagDateType.map()54        return map_[self]55def make_mag_query(56    start_date: Optional[pendulum.DateTime] = None,57    end_date: Optional[pendulum.DateTime] = None,58    state: Optional[MagState] = None,59    date_type: MagDateType = MagDateType.release,60) -> str:61    """Build a query for the MagReleases table.62    :param start_date: start date for the query (inclusive).63    :param end_date: end date for the query (exclusive).64    :param state: the state of the MAG release.65    :param date_type: the date type to query.66    :return: the query.67    """68    commands = []69    date_format = "%Y-%m-%dT%H:%MZ"70    if state is not None:71        commands.append(f"State eq '{state.value}'")72    if start_date is not None:73        commands.append(f"{date_type.value} ge datetime'{start_date.strftime(date_format)}'")74    if end_date is not None:75        commands.append(f"{date_type.value} lt datetime'{end_date.strftime(date_format)}'")76    query = " and ".join(commands)77    return query78def hide_if_not_none(secret: Any):79    """Hide a secret unless it is None, which means that the user didn't set it.80    :param secret: the secret.81    :return: None or 'hidden'82    """83    value = None84    if secret is not None:85        value = "hidden"86    return value87class MagContainer:88    def __init__(self, name: str, last_modified: pendulum.DateTime, release_date: pendulum.Date):89        """Microsoft Academic Graph container class.90        :param name: name of the container.91        :param last_modified: date that the container was last modified.92        :param release_date: date that the MAG release was released.93        """94        self.name = name95        self.last_modified = last_modified96        self.release_date = release_date97    def __str__(self):98        return f"Mag Container {self.name}"99    def __repr__(self):100        return (101            f'MagContainer({self.name}, {self.last_modified.strftime("%Y-%m-%d")}, '102            f'{self.release_date.strftime("%Y-%m-%d")})'103        )104class MagRelease:105    TABLE_NAME = "MagReleases"106    __PARTITION_KEY = "PartitionKey"107    __ROW_KEY = "RowKey"108    __STATE = "State"109    __TASK = "Task"110    __RELEASE_DATE = "ReleaseDate"111    __SOURCE_CONTAINER = "SourceContainer"112    __SOURCE_CONTAINER_LAST_MODIFIED = "SourceContainerLastModified"113    __RELEASE_CONTAINER = "ReleaseContainer"114    __RELEASE_PATH = "ReleasePath"115    __DISCOVERED_DATE = "DiscoveredDate"116    __ARCHIVED_DATE = "ArchivedDate"117    __DONE_DATE = "DoneDate"118    def __init__(119        self,120        partition_key: str,121        row_key: str,122        state: MagState,123        task: MagTask,124        release_date: pendulum.Date,125        source_container: str,126        source_container_last_modified: pendulum.DateTime,127        release_container: str,128        release_path: str,129        discovered_date: pendulum.DateTime,130        archived_date: pendulum.DateTime,131        done_date: pendulum.DateTime,132        account_name: Optional[str] = None,133        account_key: Optional[str] = None,134    ):135        """Microsoft Academic Graph release class.136        :param partition_key: partition key.137        :param row_key: the row key. partition_key + row_key form the primary key.138        :param state: state that the MAG release is in.139        :param task: current task that is being executed.140        :param release_date: the official release date for the MAG release. Note that releases are provisioned onto141        the Azure storage account about a week after the "release_date".142        :param source_container: container where the MAG release was provisioned to.143        :param source_container_last_modified: date that the source container was last modified.144        :param release_container: container where the MAG releases are staged for data transfer.145        :param release_path: path to the MAG release on the release container.146        :param discovered_date: date that the MAG release was discovered by the MAG archiver.147        :param archived_date: date the the MAG release was archived. Set to the minimum date, i.e.148        1601-01-01T00:00:00.000Z if not specified.149        :param done_date: date the MAG release is ready to be transferred. Set to the minimum date, i.e.150        1601-01-01T00:00:00.000Z if not specified.151        :param account_name: Azure Storage account name.152        :param account_key: Azure Storage account key.153        """154        self.partition_key = partition_key155        self.row_key = row_key156        self.state = state157        self.task = task158        self.release_date = release_date159        self.source_container = source_container160        self.source_container_last_modified = source_container_last_modified161        self.release_container = release_container162        self.release_path = release_path163        self.discovered_date = discovered_date164        self.archived_date = archived_date165        self.done_date = done_date166        self.account_name = account_name167        self.account_key = account_key168    def __assert_account(self, msg):169        assert self.account_name is not None and self.account_key is not None, msg170    def __table_service(self):171        return TableService(account_name=self.account_name, account_key=self.account_key)172    def create(self) -> bool:173        """Create the MagRelease instance in the MagReleases table.174        :return: whether the instance was created or not.175        """176        self.__assert_account("MagRelease.create: account_name and account_key must be supplied.")177        service = self.__table_service()178        success = True179        try:180            service.insert_entity(MagRelease.TABLE_NAME, self.to_entity())181        except AzureConflictHttpError as e:182            success = False183            logging.error(e)184        return success185    def delete(self) -> None:186        """Delete this MagRelease instance from the MagReleases table.187        :return: None.188        """189        self.__assert_account("MagRelease.delete: account_name and account_key must be supplied.")190        service = self.__table_service()191        service.delete_entity(MagRelease.TABLE_NAME, self.partition_key, self.row_key)192    def archive(self, target_container: str, target_folder: str):193        """Archive this MAG release into to the target container.194        TODO: in the future this method should compress the release before copying it.195        :param target_container: the container to copy the MAG release too.196        :param target_folder: the folder within the target container to copy the MAG release into.197        :return: None.198        """199        self.__assert_account("MagRelease.archive: account_name and account_key must be supplied.")200        return copy_container(201            self.account_name, self.account_key, self.source_container, target_container, target_folder202        )203    def cleanup(self):204        """Cleanup the source container.205        :return:206        """207        self.__assert_account("MagRelease.cleanup: account_name and account_key must be supplied.")208        return delete_container(self.account_name, self.account_key, self.source_container)209    def update(self):210        """Update this MAG release instance in the MagReleases table.211        :return:212        """213        self.__assert_account("MagRelease.update_state: account_name and account_key must be supplied.")214        # Update properties215        service = self.__table_service()216        return service.update_entity(MagRelease.TABLE_NAME, self.to_entity())217    @staticmethod218    def from_entity(entity: dict, account_name: Optional[str] = None, account_key: Optional[str] = None):219        """Create a MagRelease instance from an entity dictionary.220        :param entity: the entity returned from the Azure Table Storage API.221        :param account_name: Azure Storage account name. Optional, only required if the create, delete, archive,222        cleanup or update functions need to be used.223        :param account_key: Azure Storage account key. Optional, only required if the create, delete, archive,224        cleanup or update functions need to be used.225        :return: the MagRelease instance.226        """227        partition_key_ = entity[MagRelease.__PARTITION_KEY]228        row_key_ = entity[MagRelease.__ROW_KEY]229        state_ = MagState(entity[MagRelease.__STATE])230        task_ = MagTask(entity[MagRelease.__TASK])231        release_date_ = pendulum.instance(entity[MagRelease.__RELEASE_DATE])232        source_container_ = entity[MagRelease.__SOURCE_CONTAINER]233        source_container_last_modified_ = entity[MagRelease.__SOURCE_CONTAINER_LAST_MODIFIED]234        release_container_ = entity[MagRelease.__RELEASE_CONTAINER]235        release_path_ = entity[MagRelease.__RELEASE_PATH]236        discovered_date_ = pendulum.instance(entity[MagRelease.__DISCOVERED_DATE])237        archived_date_ = pendulum.instance(entity[MagRelease.__ARCHIVED_DATE])238        done_date_ = pendulum.instance(entity[MagRelease.__DONE_DATE])239        return MagRelease(240            partition_key_,241            row_key_,242            state_,243            task_,244            release_date_,245            source_container_,246            source_container_last_modified_,247            release_container_,248            release_path_,249            discovered_date_,250            archived_date_,251            done_date_,252            account_name=account_name,253            account_key=account_key,254        )255    def to_entity(self) -> dict:256        """Convert a MagRelease instance into an Azure Table Storage entity.257        :return: an Azure Table Storage entity.258        """259        entity = dict()260        entity[MagRelease.__PARTITION_KEY] = self.partition_key261        entity[MagRelease.__ROW_KEY] = self.row_key262        entity[MagRelease.__STATE] = self.state.value263        entity[MagRelease.__TASK] = self.task.value264        entity[MagRelease.__RELEASE_DATE] = EntityProperty(EdmType.DATETIME, value=self.release_date)265        entity[MagRelease.__SOURCE_CONTAINER] = self.source_container266        entity[MagRelease.__SOURCE_CONTAINER_LAST_MODIFIED] = EntityProperty(267            EdmType.DATETIME, value=self.source_container_last_modified268        )269        entity[MagRelease.__RELEASE_CONTAINER] = self.release_container270        entity[MagRelease.__RELEASE_PATH] = self.release_path271        entity[MagRelease.__DISCOVERED_DATE] = EntityProperty(EdmType.DATETIME, value=self.discovered_date)272        entity[MagRelease.__ARCHIVED_DATE] = EntityProperty(EdmType.DATETIME, value=self.archived_date)273        entity[MagRelease.__DONE_DATE] = EntityProperty(EdmType.DATETIME, value=self.done_date)274        return entity275    def __str__(self):276        return f'MagRelease {self.release_date.strftime("%Y-%m-%d")}'277    def __repr__(self):278        dt_format = "%Y-%m-%dT%H:%MZ"279        return (280            "MagRelease("281            f"{self.partition_key}, "282            f"{self.row_key}, "283            f"{self.state.value}, "284            f"{self.task.value}, "285            f"{self.release_date.strftime(dt_format)}, "286            f"{self.source_container}, "287            f"{self.source_container_last_modified.strftime(dt_format)}, "288            f"{self.release_container}, "289            f"{self.release_path}, "290            f"{self.discovered_date.strftime(dt_format)}, "291            f"{self.archived_date.strftime(dt_format)}, "292            f"{self.done_date.strftime(dt_format)}, "293            f"account_name={self.account_name}, "294            f"account_key={hide_if_not_none(self.account_key)})"295        )296class MagArchiverClient:297    __MAG_RELEASE_RE = re.compile("mag-[0-9]{4}-[0-9]{2}-[0-9]{2}")298    def __init__(299        self, account_name: Optional[str] = None, account_key: Optional[str] = None, sas_token: Optional[str] = None300    ):301        """302        :param account_name: the name of the Azure Storage account where the Microsoft Academic Graph releases are303        provisioned and the `MagReleases` Azure Storage Table is stored.304        :param account_key: the key to Azure Storage account.305        :param sas_token: the shared access token to use for authentication instead of the account_key.306        """307        self.account_name = account_name308        self.account_key = account_key309        self.sas_token = sas_token310    def list_containers(311        self, last_modified_thresh: Optional[float] = None, reverse: bool = False312    ) -> List[MagContainer]:313        """List all Azure Storage Blob Containers holding MAG releases.314        :param last_modified_thresh: only include containers that were last modified greater than or equal to a315        specific number of hours.316        :param reverse: sort from oldest to newest using release_date datetime.317        :return: the containers holding MAG releases.318        """319        # List all containers in the storage account320        containers: List[ContainerProperties] = list_containers(self.account_name, self.account_key)321        tz = pendulum.timezone("UTC")322        current_time = pendulum.now(tz)323        # Select all containers containing MAG releases that were last updated greater than324        mag_containers = []325        for container in containers:326            container_name = container.name327            last_modified = pendulum.instance(container.last_modified, tz=tz)328            hours_since_modified = current_time.diff(last_modified).in_hours()329            # Only include containers with a name that matches the MAG container pattern330            if MagArchiverClient.__MAG_RELEASE_RE.match(container_name) is not None:331                # Add of no last_modified_thresh supplied or if the hours since modified is greater than the last332                # modified threshold333                if last_modified_thresh is None or last_modified_thresh <= hours_since_modified:334                    release_date_str = container.name.replace("mag-", "")335                    release_date = pendulum.parse(release_date_str)336                    mag_container = MagContainer(container_name, last_modified, release_date)337                    mag_containers.append(mag_container)338        # Sort from oldest to newest339        mag_containers.sort(key=lambda c: c.release_date, reverse=reverse)340        return mag_containers341    def update_releases(self, containers: List[MagContainer]):342        """Update the releases in the MagReleases table based on a list of containers containing MAG releases.343        :param containers: the containers containing MAG releases.344        :return: the number of releases created and the number of errors.345        """346        min_date = pendulum.datetime(1601, 1, 1)347        discovered_date = pendulum.now()348        # Get all entities349        table_service = TableService(account_name=self.account_name, account_key=self.account_key)350        entities = table_service.query_entities(MagRelease.TABLE_NAME)351        # Get all containers that are not in the MagReleases table352        new_container_index = dict.fromkeys(353            set([container.name for container in containers]) - set([entity["SourceContainer"] for entity in entities]),354            0,355        )356        num_new_containers = len(new_container_index)357        logging.info(f"Num new containers discovered: {num_new_containers}")358        # Only add new containers359        num_created = 0360        num_errors = 0361        for container in containers:362            if container.name in new_container_index:363                partition_key = "mag"364                row_key = container.release_date.strftime("%Y-%m-%d")365                release = MagRelease(366                    partition_key,367                    row_key,368                    MagState.discovered,369                    MagTask.not_started,370                    container.release_date,371                    container.name,372                    container.last_modified,373                    "",374                    "",375                    discovered_date,376                    min_date,377                    min_date,378                    account_name=self.account_name,379                    account_key=self.account_key,380                )381                success = release.create()382                if success:383                    num_created += 1384                else:385                    num_errors += 1386        return num_created, num_errors387    def list_releases(388        self,389        start_date: Optional[pendulum.DateTime] = None,390        end_date: Optional[pendulum.DateTime] = None,391        state: Optional[MagState] = None,392        date_type: MagDateType = MagDateType.release,393        reverse: bool = False,394    ) -> List[MagRelease]:395        """List Microsoft Academic releases in the MagReleases Azure Storage Table.396        :param start_date: start date for the query.397        :param end_date: end date for the query.398        :param state: the state of the MAG release.399        :param date_type: the date type to query.400        :param reverse: whether to reverse sort the results. Sorting is performed on the date type.401        :return: the list of MagReleases found.402        """403        # Query and fetch releases404        table_service = TableService(405            account_name=self.account_name, account_key=self.account_key, sas_token=self.sas_token406        )407        query = make_mag_query(start_date=start_date, end_date=end_date, state=state, date_type=date_type)408        entities = table_service.query_entities("MagReleases", filter=query)409        # Convert entities into MagRelease objects410        releases = []411        for entity in entities:412            release = MagRelease.from_entity(entity, account_name=self.account_name, account_key=self.account_key)413            releases.append(release)414        # Sort from oldest to newest, unless reverse is set415        releases.sort(key=lambda r: getattr(r, MagDateType.attr(date_type)), reverse=reverse)416        return releases417    def __str__(self):418        return self.__repr__()419    def __repr__(self):420        return (421            "MagArchiverClient("422            f"account_name={self.account_name}, "423            f"account_key={hide_if_not_none(self.account_key)}, "424            f"sas_token={hide_if_not_none(self.sas_token)})"...

releaseTimeRollup.py

Source:releaseTimeRollup.py

1import argparse2import epicTimeRollup3from dataclasses import dataclass, asdict4import os5import shutil6from jira import JIRA7import json8from subprocess import Popen9@dataclass10class Release:11    release: str12    issues: []13    summed_time: float14    def dict(self):15        issues_json = []16        est_count = 017        for issue in self.issues:18            di = issue.dict()19            issues_json.append(issue.dict())20            if issue.summed_time > 0.0:21                est_count += 122        return {'key': self.release.replace(" ", "_"), 'time': self.summed_time, 'subticket_count': len(self.issues), 'subticket_estimate_count': est_count, 'issues': issues_json}23def export_releases_json(root, releases_container):24    """25        file - fully qualified path to file in which to write to.26        releasess_container - the list of release DataObjects to export as JSON.27    """28    releases_json = {}29    for release_container in releases_container:30        if release_container.release not in releases_json:31            releases_json[release_container.release] = []32        print("Processing to JSON structure of {}".format(33            release_container.release))34        releases_json[release_container.release].append(35            release_container.dict())36    for release_key in releases_json:37        out_file_path = os.path.join(38            root, "{}_estimates.json".format(release_key.replace(" ", "_")))39        with open(out_file_path, "w") as output_file:40            output_file.writelines(json.dumps(41                releases_json[release_key], indent=4, separators=(",", ": ")))42    print("Finished writing to file.")43def parse_args(args_list):44    """45    Parse arguments for release-time-rollup.46    """47    parser = argparse.ArgumentParser()48    parser.add_argument("--user", required=True)49    parser.add_argument("--api_token", required=True)50    parser.add_argument("--releases", required=True)51    parser.add_argument("--export_estimates", action='store_true')52    parser.add_argument("--export_estimates_path")53    parser.add_argument("--export_project_configs", action='store_true')54    parser.add_argument("--export_project_config_path")55    parser.add_argument("--import_project_configs", action='store_true')56    parser.add_argument("--import_project_configs_path")57    args = parser.parse_args(args=args_list)58    return args59def execute(args_list):60    args = parse_args(args_list)61    print("Running JIRA Tabulations for Releases")62    jira_options = {"server": "https://battlefy.atlassian.net"}63    jira = JIRA(64        options=jira_options,65        basic_auth=(args.user, args.api_token),66    )67    releases = args.releases.split(",")68    releases_container = []69    project_configs = {}70    for release in releases:71        release_obj = Release(release, [], 0.0)72        releases_container.append(release_obj)73        query_string = "fixVersion={}".format(release)74        # get a list of the issues first, just by summary and comprehend the75        # projects76        issue_projects = list(set([e.fields.project.id for e in jira.search_issues(77            query_string)]))78        if len(issue_projects) != 1:79            print("Multiple projects in release; unable to assert size.")80            return -181        root_project_id = issue_projects[0]82        if root_project_id not in project_configs:83            project_configs[root_project_id] = epicTimeRollup.generate_project_constants(84                jira, jira.project(root_project_id))85        cust_keys = list(set([project_configs[root_project_id].story.estimation_key,86                              project_configs[root_project_id].task.estimation_key]))87        cust_key_str = ",".join(cust_keys)88        release_obj.issues = [epicTimeRollup.UserStory(89            e, e.fields.subtasks if hasattr(90                e.fields, "subtasks") else [], 0.091        )92            for e in jira.search_issues(93            query_string, fields="{}, subtasks, summary, issuetype".format(cust_key_str))94        ]95        for release in releases_container:96            for issue in release.issues:97                epicTimeRollup.extract_issue_estimate(98                    jira, issue, project_configs[root_project_id])99                release.summed_time += issue.summed_time100    if args.export_estimates:101        export_releases_json(...

cgroup_manager.py

Source:cgroup_manager.py

...38    @abstractmethod39    def get_memory_spread_slab(self, container_name) -> bool:40        pass41    @abstractmethod42    def release_container(self, container_name):43        pass44    @abstractmethod45    def get_isolated_workload_ids(self) -> List[str]:46        pass47    @abstractmethod48    def has_pending_work(self) -> bool:...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.