KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run autotest automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
