KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use commit method in localstack
Best Python code snippet using localstack_python
rit_lib.py
Source:rit_lib.py 
...192 self._branch_names = cleared_rit._branch_names193 self._commit_ids = cleared_rit._commit_ids194 self._short_commit_tree = cleared_rit._short_commit_tree195 ''' SET '''196 def add_commit(self, commit: Commit):197 ''' add the commit to the rit dir '''198 if self.prevent_mutations:199 raise RitError("Doing this would mutate the rit directory, and that is disabled for this RitResource")200 self._write_commit(commit)201 self._clear()202 def prune(self):203 leaf_commits = set()204 if self.head.commit_id is not None:205 leaf_commits.add(self.head.commit_id)206 for branch_name in self.get_branch_names():207 leaf_commits.add(self.get_branch(branch_name, ensure=True).commit_id)208 refed_commits = set(leaf_commits)209 for leaf_commit in leaf_commits:210 while True:211 commit = self.get_commit(leaf_commit, ensure=True)212 if commit.parent_commit_id is None or commit.parent_commit_id in refed_commits:213 break214 refed_commits.add(commit.parent_commit_id)215 leaf_commit = commit.parent_commit_id216 removed_commit_ids = []217 for commit_id in self.get_commit_ids():218 if commit_id in refed_commits:219 continue220 self._delete_commit(commit_id)221 removed_commit_ids.append(commit_id)222 return removed_commit_ids223 def set_branch(self, branch: Branch):224 ''' add the branch to the rit dir '''225 if self.prevent_mutations:226 raise RitError("Doing this would mutate the rit directory, and that is disabled for this RitResource")227 self._write_branch(branch)228 self._clear()229 def set_head(self, head: HeadNode):230 ''' set the new head point '''231 if self.prevent_mutations:232 raise RitError("Doing this would mutate the rit directory, and that is disabled for this RitResource")233 self._write_head(head)234 self._clear()235 ''' GET '''236 @property237 def paths(self):238 ''' returns a RitPaths object for this rit directory '''239 if self._paths is None:240 self._paths = self._read_paths()241 return self._paths242 @property243 def head(self):244 ''' returns the HeadNode object for this rit directory '''245 if self._head is None:246 self._head = self._read_head()247 return self._head248 def get_head_commit_id(self):249 ''' get the commit id of the head, None if there isn't one '''250 head = self.head251 if head.commit_id is not None:252 return head.commit_id253 else:254 branch = self.get_branch(head.branch_name)255 if branch is not None:256 return branch.commit_id257 else:258 return None259 def get_commit_ids(self):260 ''' get all commit ids '''261 if self._commit_ids is None:262 self._commit_ids = self._read_commit_ids()263 return self._commit_ids264 def get_commit(self, commit_id: str, *, ensure=False):265 '''266 get commit object of a commit id267 return a commit if found268 otherwise return None, if ensure is set to True, the raise instead of return None.269 '''270 if commit_id not in self._commits:271 try:272 self._commits[commit_id] = self._read_commit(commit_id)273 except FileNotFoundError:274 if ensure:275 raise RitError("Unable to load expected commit")276 return None277 return self._commits[commit_id]278 def is_commit(self, commit_id: str):279 ''' return True if commit_id has a commit '''280 return self.get_commit(commit_id) is not None281 def get_branch_names(self):282 ''' return names of all branches '''283 if self._branch_names is None:284 self._branch_names = self._read_branch_names()285 return self._branch_names286 def get_branch(self, name: str, *, ensure=False):287 '''288 query for branch by name289 returns None if not found. raises RitError instead if ensure is True290 '''291 if name not in self._branches:292 try:293 self._branches[name] = self._read_branch(name)294 except FileNotFoundError:295 if ensure:296 raise RitError("Unable to load expected branch")297 return None298 return self._branches[name]299 def is_branch(self, name: str):300 ''' return True if the branch name has a branch '''301 return self.get_branch(name) is not None302 def get_branch_name_to_commit_ids(self):303 ''' see _populate_commit_to_branch_map '''304 if self._branch_name_to_commit_ids is None:305 self._populate_commit_to_branch_map()306 return self._branch_name_to_commit_ids307 def get_commit_id_to_branch_names(self):308 ''' see _populate_commit_to_branch_map '''309 if self._commit_id_to_branch_names is None:310 self._populate_commit_to_branch_map()311 return self._commit_id_to_branch_names312 def get_commit_tree(self):313 '''314 returns a map of shortened commit prefixes to a list of all commit ids with315 that prefix316 the tree is used to find full commit ids given partial ones.317 '''318 if self._short_commit_tree is None:319 self._short_commit_tree = defaultdict(list)320 for commit_id in self.get_commit_ids():321 self._short_commit_tree[commit_id[:short_hash_index]].append(commit_id)322 return self._short_commit_tree323 ''' helpers '''324 def _populate_commit_to_branch_map(self):325 '''326 populate the commit <-> branch maps327 branches include the HEAD node328 if a branch doesn't have a commit, the commit and branch will not have329 entries in the maps330 '''331 branch_names = self.get_branch_names()332 self._branch_name_to_commit_ids = {}333 self._commit_id_to_branch_names = defaultdict(list)334 for branch_name in branch_names:335 branch = self.get_branch(branch_name, ensure=True)336 commit_id = branch.commit_id337 self._branch_name_to_commit_ids[branch_name] = commit_id338 self._commit_id_to_branch_names[commit_id].append(branch_name)339 head_commit_id = self.get_head_commit_id()340 if head_commit_id is not None:341 self._branch_name_to_commit_ids[head_ref_name] = head_commit_id342 self._commit_id_to_branch_names[head_commit_id].append(head_ref_name)343 ''' IO '''344 def _read_paths(self):345 root_rit_dir = os.path.realpath(self.root_rit_dir)346 rit_dir = os.path.join(root_rit_dir, rit_dir_name)347 last_root_rit_dir = None348 while not os.path.isdir(rit_dir):349 last_root_rit_dir = root_rit_dir350 root_rit_dir = os.path.dirname(root_rit_dir)351 if last_root_rit_dir == root_rit_dir:352 raise RitError("Unable to locate rit directory")353 rit_dir = os.path.join(root_rit_dir, rit_dir_name)354 return RitPaths.build_rit_paths(root_rit_dir)355 def _read_head(self):356 try:357 with open(os.path.join(self.paths.rit_dir, head_ref_name)) as fin:358 return HeadNode(**json.load(fin))359 except FileNotFoundError:360 return HeadNode(None, default_branch_name)361 def _read_commit(self, commit_id: str):362 logger.debug("Reading commit: %s", commit_id)363 with open(os.path.join(self.paths.commits, commit_id)) as fin:364 return Commit(**dict(**json.load(fin), commit_id=commit_id))365 def _write_commit(self, commit: Commit):366 logger.debug("Writing commit: %s", commit.commit_id)367 with open(os.path.join(self.paths.commits, commit.commit_id), 'w') as fout:368 data = asdict(commit)369 del data['commit_id']370 json.dump(data, fout)371 def _delete_commit(self, commit_id: str):372 try:373 os.remove(os.path.join(self.paths.commits, commit_id))374 except FileNotFoundError:375 raise RitError("Failed to remove commit since it didn't exist: %s", commit_id)376 try:377 os.remove(get_tar_path(self, commit_id))378 except FileNotFoundError:379 raise RitError("Failed to remove commit's tar since it didn't exist: %s", commit_id)380 try:381 os.remove(get_snar_path(self, commit_id))382 except FileNotFoundError:383 raise RitError("Failed to remove commit's snar since it didn't exist: %s", commit_id)384 def _read_branch(self, name: str):385 logger.debug("Reading branch: %s", name)386 with open(os.path.join(self.paths.branches, name)) as fin:387 branch = json.load(fin)388 return Branch(**dict(**branch, name=name))389 def _write_branch(self, branch: Branch):390 logger.debug("Writing branch %s to %s", branch.name, branch.commit_id)391 if self.is_commit(branch.name):392 raise RitError('Not creating a branch with the same name as a commit id: %s', branch.name)393 data = asdict(branch)394 del data['name']395 with open(os.path.join(self.paths.branches, branch.name), 'w') as fout:396 json.dump(data, fout)397 def _read_branch_names(self):398 for _, _, branch_names in os.walk(self.paths.branches):399 return branch_names400 return []401 def _read_commit_ids(self):402 for _, _, commit_ids in os.walk(self.paths.commits):403 return commit_ids404 return []405 def _write_head(self, head: HeadNode):406 with open(os.path.join(self.paths.rit_dir, head_ref_name), 'w') as fout:407 json.dump(asdict(head), fout)408''' UTIL '''409def colorize(color: int, msg: str):410 reset_seq = "\033[0m"411 color_seq = "\033[1;{}m"412 color_section = color_seq + "{}" + reset_seq413 return color_section.format(color, msg)414def mkdir(*args, exists_ok=False, **kwargs):415 try:416 os.mkdir(*args, **kwargs)417 except FileExistsError:418 if not exists_ok:419 raise420def none_t():421 def none_type(obj):422 return obj is None423 return none_type424def exact_t(*types):425 def exact_type(obj):426 return isinstance(obj, types)427 return exact_type428def optional_t(obj_t):429 def optional_type(obj):430 if obj is None:431 return True432 else:433 return obj_t(obj)434 return optional_type435def list_t(obj_t):436 def list_type(obj):437 if isinstance(obj, list):438 return all(obj_t(val) for val in obj)439 return False440 return list_type441def check_types(**type_defs):442 for name, (obj, type_def) in type_defs.items():443 if not type_def(obj):444 raise TypeError(f"Element had invalid type: {name}: {type(obj)}")445def check_obj_types(obj, type_defs):446 objs = {}447 for key, type_def in type_defs.items():448 objs[key] = (getattr(obj, key), type_def)449def require(statement, msg, *args):450 if not statement:451 raise RitError(msg, *args)452''' RIT DIR HELPERS '''453def get_tar_path(rit: RitResource, commit_id: str):454 ''' get a commit's tar path, where the backup for that commit is '''455 return os.path.join(rit.paths.backups, commit_id + '.tar')456def get_snar_path(rit: RitResource, commit_id: str):457 ''' get a commit's star path, where the backup's metadata for that commit is '''458 return os.path.join(rit.paths.backups, commit_id + '.snar')459''' COMMIT HELPERS '''460def hash_commit(create_time: float, msg: str, snar: str, tar: str):461 ''' create a commit_id from '''462 logger.debug("Calculating the hash of ref")463 ref_hash = hashlib.sha1()464 ref_hash.update(b'create_time')465 ref_hash.update(str(create_time).encode('utf-8'))466 ref_hash.update(b'msg')467 ref_hash.update(msg.encode('utf-8'))468 ref_hash.update(b'snar')469 with open(snar, 'rb') as fin:470 ref_hash.update(fin.read())471 ref_hash.update(b'tar')472 with open(tar, 'rb') as fin:473 ref_hash.update(fin.read())474 return ref_hash.hexdigest()475def check_tar():476 ''' verify tar is the correct version and GNU '''477 logger.debug("Checking tar version")478 process = subprocess.Popen(['tar', '--version'], stdout=subprocess.PIPE)479 contents = process.stdout.read()480 process.wait()481 version = contents.decode('utf-8').split('\n', 1)[0]482 logger.debug("Tar Version: %s", version)483 assert 'GNU tar' in version, "You must have a GNU tar installed"484def status_tar(rit: RitResource, verbose: bool):485 ''' returns True if rit directory is dirty '''486 parent_commit_id = rit.get_head_commit_id()487 work_snar = os.path.join(rit.paths.work, 'ref.snar')488 if parent_commit_id is not None:489 head_snar = get_snar_path(rit, parent_commit_id)490 # TODO: move into rit resource491 shutil.copyfile(head_snar, work_snar)492 check_tar()493 tar_cmd = ['tar', '-cvg', work_snar, f'--exclude={rit_dir_name}', '-f', os.devnull, '.']494 logger.debug("Running tar command: %s", tar_cmd)495 # TODO: move into rit resource496 process = subprocess.Popen(tar_cmd, cwd=rit.paths.root, stdout=subprocess.PIPE)497 terminated = False498 dirty = False499 while True:500 line = process.stdout.readline()501 if not line:502 break503 if line == b'./\n':504 continue505 dirty = True506 if verbose:507 output = line.decode('utf-8').strip()508 logger.info("\t- %s", colorize(fg + red, output))509 else:510 terminated = True511 try:512 process.terminate()513 except Exception:514 pass515 break516 # still need to read the stdout to prevent blocking and therefore deadlock517 if terminated:518 while process.stdout.read(DEFAULT_BUFFER_SIZE):519 pass520 exit_code = process.wait()521 if not terminated and exit_code != 0:522 raise RitError("Creating commit's tar failed with exit code: %d", exit_code)523 os.remove(work_snar)524 return dirty525def create_commit(rit: RitResource, create_time: float, msg: str):526 ''' create a commit with the current head as the parent commit (if any) '''527 parent_commit_id = rit.get_head_commit_id()528 logger.debug("Parent ref: %s", parent_commit_id)529 work_tar = os.path.join(rit.paths.work, 'ref.tar')530 logger.debug("Working tar: %s", work_tar)531 work_snar = os.path.join(rit.paths.work, 'ref.snar')532 logger.debug("Working snar: %s", work_snar)533 if parent_commit_id is not None:534 head_snar = get_snar_path(rit, parent_commit_id)535 logger.debug("Copying previous snar: %s", head_snar)536 # TODO: move into rit resource537 shutil.copyfile(head_snar, work_snar)538 else:539 logger.debug("Using fresh snar file since no parent commit")540 check_tar()541 opts = '-cz'542 if logger.getEffectiveLevel() <= logging.DEBUG:543 opts += 'v'544 opts += 'g'545 tar_cmd = ['tar', opts, work_snar, f'--exclude={rit_dir_name}', '-f', work_tar, '.']546 logger.debug("Running tar command: %s", tar_cmd)547 # TODO: move into rit resource548 process = subprocess.Popen(tar_cmd, cwd=rit.paths.root)549 # TODO: doesn't forward SIGTERM, only SIGINT550 exit_code = process.wait()551 if exit_code != 0:552 raise RitError("Creating commit's tar failed with exit code: %d", exit_code)553 commit_id = hash_commit(create_time, msg, work_snar, work_tar)554 logger.debug("Moving working snar into backups directory")555 snar = get_snar_path(rit, commit_id)556 os.rename(work_snar, snar)557 logger.debug("Moving working tar into backups directory")558 tar = get_tar_path(rit, commit_id)559 os.rename(work_tar, tar)560 commit = Commit(parent_commit_id, commit_id, create_time, msg)561 rit.add_commit(commit)562 if rit.head.commit_id is not None:563 new_head = HeadNode(commit_id=commit_id, branch_name=None)564 rit.set_head(new_head)565 else:566 rit.set_branch(Branch(rit.head.branch_name, commit_id))567 return commit568''' RESET HELPERS '''569def apply_commit(rit: RitResource, commit: Commit):570 '''571 apply the commit to the rit directory572 If the current rit directory isn't clean and isn't the parent of the commit573 being applied, the results will not be the contents of commit.574 See restore_to_commit.575 '''576 logger.info("Applying commit: %s", commit.commit_id)577 tar_file = get_tar_path(rit, commit.commit_id)578 tar_cmd = ['tar', '-xg', os.devnull, '-f', tar_file]579 # rit resource thing?580 process = subprocess.Popen(tar_cmd, cwd=rit.paths.root)581 exit_code = process.wait()582 if exit_code != 0:583 raise RitError("Failed while trying to apply commit: %s", commit.commit_id)584def restore_to_commit(rit: RitResource, commit: Commit):585 '''586 This gets the chain of commits from commit to root and applies them. If there587 are changes in the working directory relative to current head, then those will588 be destroyed.589 '''590 logger.debug('resetting to %s', commit.commit_id)591 commit_chain = [commit]592 while commit.parent_commit_id is not None:593 commit = rit.get_commit(commit.parent_commit_id, ensure=True)594 commit_chain.append(commit)595 commit_chain.reverse()596 for commit in commit_chain:597 apply_commit(rit, commit)598''' BRANCH HELPERS '''599@dataclass600class ResolvedRef:601 '''602 the user provides a ref, which can reference head, a branch or commit. this603 object contains the HeadNode, Branch and Commit that the user's ref is604 referring to. all 3 or None can be defined.605 see resolve_ref606 '''607 commit: Optional[Commit] = None608 '''609 if None:610 if head:611 head points to a branch with no commit612 else:613 ref doesn't refer to a branch or commit614 else:615 ref ultimately refers to this commit616 '''617 branch: Optional[Branch] = None618 '''619 if head is None620 if branch is None:621 ref doesn't refer to a branch622 else:623 ref points to this branch624 else:625 if branch is None:626 head points to a commit.627 head points to a branch with no commit.628 else:629 head points to this branch630 '''631 head: Optional[HeadNode] = None632 '''633 if head is None:634 ref was provided and not the head635 else:636 ref was omitted or explicitly set to the head637 '''638def resolve_commit(rit: RitResource, partial_commit_id: str):639 '''640 resolve a user provided commit id to a commit. if no commit is found, then641 return None. if the ref is an ambiguous shortened commit id, then this642 function raises an exception.643 '''644 logger.debug("Resolving commit: %s", partial_commit_id)645 commit = rit.get_commit(partial_commit_id)646 if commit is not None:647 return commit648 if len(partial_commit_id) < short_hash_index:649 return None650 short_commit_id = partial_commit_id[:short_hash_index]651 commit_tree = rit.get_commit_tree()652 if short_commit_id not in commit_tree:653 return None654 commit = None655 for commit_id in commit_tree[short_commit_id]:656 size = len(partial_commit_id)657 if partial_commit_id == commit_id[:size]:658 if commit is not None:659 raise RitError("Reference %s matched commits %s and %s", partial_commit_id, commit.commit_id, commit_id)660 commit = rit.get_commit(commit_id, ensure=True)661 return commit662def resolve_ref(rit: RitResource, ref: Optional[str]):663 '''664 resolve a user provided reference665 if ref is None, then ref refers to the current head. ref can also explicitly666 refer to the current head. ref otherwise refers to a branch. if the branch is667 not found, then the ref refers to a commit. if no commit is found, all 3668 fields of ResolvedRef will be None.669 if the ref is an ambiguous shortened commit id, then this function raises an670 exception.671 see the def of ResolvedRef672 '''673 logger.debug("Resolving ref: %s", ref)674 res = ResolvedRef()675 if ref is None or ref == head_ref_name:676 head = rit.head677 res.head = head678 if head.branch_name is not None:679 res.branch = rit.get_branch(head.branch_name)680 if res.branch is not None:681 res.commit = rit.get_commit(res.branch.commit_id)682 else:683 res.commit = rit.get_commit(head.commit_id)684 else:685 res.branch = rit.get_branch(ref)686 if res.branch is not None:687 res.commit = rit.get_commit(res.branch.commit_id)688 else:689 res.commit = resolve_commit(rit, ref)690 return res691def resolve_refs(rit: RitResource, refs: list[str], all: bool):692 '''693 Returns information regarding the provided refs694 '''695 resolved_refs: list[ResolvedRef] = []696 if not refs:697 refs.append(None)698 if all:699 refs.extend(rit.get_branch_names())700 for ref in refs:701 res = resolve_ref(rit, ref)702 resolved_refs.append(res)703 return resolved_refs704branch_name_re = re.compile('^\\w+$')705def validate_branch_name(name: str):706 ''' return whether this string is a valid branch name '''707 if name == head_ref_name:708 raise RitError("Branch can't be named the same as the head ref: %s", name)709 elif branch_name_re.search(name) is None:710 raise RitError("Invalid branch name: %s", name)711def _pprint_dur(dur: int, name: str):712 return f"{dur} {name}{'s' if dur > 1 else ''}"713def pprint_time_duration(start: float, end: float):714 ''' pretty print a time duration '''715 start_dt = datetime.datetime.fromtimestamp(start)716 end_dt = datetime.datetime.fromtimestamp(end)717 dur = end - start718 dur_sec = dur719 dur_min = dur / 60720 dur_hour = dur_min / 60721 dur_day = dur_hour / 24722 dur_month = 12 * (end_dt.year - start_dt.year) + (end_dt.month - start_dt.month)723 dur_year = dur_month // 12724 parts = []725 if dur_year >= 5:726 parts.append(_pprint_dur(int(dur_year), 'year'))727 elif dur_year >= 1:728 parts.append(_pprint_dur(int(dur_year), 'year'))729 parts.append(_pprint_dur(int(dur_month) % 12, 'month'))730 elif dur_month >= 1:731 parts.append(_pprint_dur(int(dur_month) % 12, 'month'))732 elif dur_day >= 1:733 parts.append(_pprint_dur(int(dur_day), 'day'))734 elif dur_hour >= 1:735 parts.append(_pprint_dur(int(dur_hour) % 60, 'hour'))736 elif dur_min >= 1:737 parts.append(_pprint_dur(int(dur_min) % 60, 'minute'))738 elif dur_sec >= 20:739 parts.append(_pprint_dur(int(dur_sec) % 60, 'second'))740 else:741 return 'Just now'742 return ', '.join(parts) + ' ago'743''' SUB LOG COMMANDS '''744def log_commits(rit: RitResource, commits: list[Commit]):745 '''746 Returns tuple of the following:747 - commit_graph: a tree containing all provided commits748 - leafs: the set of leaf notes of the tree749 - commit_id_to_commit: a map of each commit to a full Commit object750 - commit_id_to_branch_names: a map of each commit_id to branch names, including the head_ref_name751 '''752 leafs: set[str] = set()753 commit_graph: dict[str, str] = {}754 for commit in commits:755 if commit.commit_id not in commit_graph:756 leafs.add(commit.commit_id)757 while True:758 commit_graph[commit.commit_id] = commit.parent_commit_id759 if commit.parent_commit_id is None:760 break761 parent_commit = rit.get_commit(commit.parent_commit_id, ensure=True)762 if parent_commit.commit_id in leafs:763 leafs.remove(parent_commit.commit_id)764 commit = parent_commit765 now = time.time()766 commit_id_to_branch_names = rit.get_commit_id_to_branch_names()767 commit_id_to_commit: dict[str, Commit] = {}768 for commit_id in leafs:769 logger.info("Log branch from %s", commit_id[:short_hash_index])770 while commit_id is not None:771 if commit_id not in commit_id_to_commit:772 commit_id_to_commit[commit_id] = rit.get_commit(commit_id, ensure=True)773 commit = commit_id_to_commit[commit_id]774 colored_commit_id = colorize(fg + yellow, commit.commit_id[:short_hash_index])775 if commit.commit_id in commit_id_to_branch_names:776 branch_names = commit_id_to_branch_names[commit.commit_id]777 colored_branch_names = []778 for branch_name in branch_names:779 if branch_name == head_ref_name:780 colored_branch_names.append(colorize(fg + blue, branch_name))781 else:782 colored_branch_names.append(colorize(fg + green, branch_name))783 branch_details = f"({', '.join(colored_branch_names)}) "784 else:785 branch_details = ''786 time_duration = pprint_time_duration(commit.create_time, now)787 date_details = f'({time_duration}) '788 logger.info("* %s %s%s%s", colored_commit_id, date_details, branch_details, commit.msg)789 commit_id = commit_graph[commit_id]790 return commit_graph, leafs, commit_id_to_commit, commit_id_to_branch_names791''' SUB BRANCH COMMANDS '''792def delete_branch(rit: RitResource, name: str):793 ''' removes a branch '''794 try:795 os.remove(os.path.join(rit.paths.branches, name))796 except FileNotFoundError:797 raise RitError("Failed to remove branch since it didn't exist.")798def list_branches(rit: RitResource):799 ''' logs branches to logger '''800 head = rit.head801 head_branch_name = head.branch_name802 branch_names = rit.get_branch_names()803 # TODO: this doesn't handle HEAD well if its commitless804 for branch_name in branch_names:805 this_sym = '*' if branch_name == head_branch_name else ' '806 branch = rit.get_branch(branch_name, ensure=True)807 commit = rit.get_commit(branch.commit_id, ensure=True)808 colored_commit_id = colorize(fg + yellow, branch.commit_id[:short_hash_index])809 colored_branch_name = colorize(fg + green, branch_name)810 logger.info("%s %s\t%s %s", this_sym, colored_branch_name, colored_commit_id, commit.msg)811 return head, branch_names812def create_branch(rit: RitResource, name: str, ref: Optional[str], force: bool):813 '''814 creates a branch with name name at ref ref. if the branch already exists,815 force will move the branch to the new commit.816 '''817 if rit.is_branch(name) and not force:818 raise RitError('Branch already exists: %s. Use -f to force the overwrite of it.', name)819 res = resolve_ref(rit, ref)820 if res.commit is None:821 if res.head is not None:822 raise RitError("Current head doesn't have a commit")823 else:824 raise RitError("Unable to resolve ref: %s", ref)825 commit_id = res.commit.commit_id826 branch = Branch(name, commit_id)827 rit.set_branch(branch)828 logger.info("Created branch %s at %s", name, commit_id[:short_hash_index])829 return res830def log_refs(rit: RitResource, refs: list[str], all: bool, full: bool):831 '''832 Returns the commit tree containing the given refs. If none are given, assumes833 head. If all is true, appends refs.834 '''835 commits = []836 if not refs:837 refs.append(None)838 if all:839 refs.extend(rit.get_branch_names())840 for ref in refs:841 res = resolve_ref(rit, ref)842 if res.commit is None:843 if res.head is not None:844 # TODO: this doesn't handle HEAD well if its commitless845 raise RitError("head branch doesn't have any commits")846 else:847 raise RitError("Unable to locate ref: %s", ref)848 commits.append(res.commit)849 return log_commits(rit, commits)850def show_ref(rit: RitResource, ref: Optional[str]):851 ''' log the contents of a specific reference '''852 res = resolve_ref(rit, ref)853 if res.commit is None:854 if res.head is not None:855 raise RitError("head branch doesn't have any commits to show")856 else:857 raise RitError("Unable to locate ref: %s", ref)858 tar_file = get_tar_path(rit, res.commit.commit_id)859 tar_cmd = ['tar', '-tf', tar_file]860 process = subprocess.Popen(tar_cmd, stdout=subprocess.PIPE)861 changes = []862 while True:863 line = process.stdout.readline()864 if not line:865 break866 elif line == b'./\n':867 continue868 output = line.decode('utf-8').strip()869 changes.append(output)870 logger.info("\t- %s", colorize(fg + cyan, output))871 results = process.wait()872 if results != 0:873 raise RitError("tar command failed with exit code %d", results)874 return res, changes875def status_head(rit: RitResource):876 ''' run status_tar on HEAD with logging '''877 if rit.head.branch_name is not None:878 head_id = rit.head.branch_name879 else:880 head_id = rit.head.commit_id881 logger.info("%s -> %s", head_ref_name, head_id)882 if not status_tar(rit, True):883 logger.info("Clean working directory!")884def prune_commits(rit: RitResource):885 ''' remove any commits not associated to a branch history '''886 return rit.prune()887def checkout_ref(*, root_rit_dir: str, ref: str, force: bool):888 '''889 Checkout ref (overwriting any changes if force is True).890 It will also check that the current working directory is clean so that we know891 no data is lost. If the working tree is dirty, then force needs to be set to892 True to forcibly delete these changes.893 Returns ref resolved, as a ResolvedRef.894 The end result is that head points to ref, and the working directory is895 restored to the point of ref.896 It is the equivalent of the following:897 - if head is a branch, set head to the underlying commit898 - hard reset to ref899 - set head to ref900 Or if you allow head to be moved to non equivalent commits:901 - set head to ref902 - hard reset to new head903 '''904 logger.debug('checkout_ref')905 logger.debug(' ref: %s', ref)906 logger.debug(' force: %s', force)907 check_types(908 ref = (ref, exact_t(str)),909 force = (force, exact_t(bool)),910 )911 rit = RitResource(root_rit_dir)912 res = resolve_ref(rit, ref)913 if res.head is not None:914 raise RitError("Attempted to checkout head ref")915 elif res.commit is None:916 raise RitError("Unable to resolve ref to commit: %s", ref)917 commit = res.commit918 restored = False919 head_commit_id = rit.get_head_commit_id()920 if head_commit_id is not None and head_commit_id != commit.commit_id:921 if not force and status_tar(rit, False):922 raise RitError("Uncommitted changes! Commit them or use -f to destroy them.")923 restored = True924 restore_to_commit(rit, commit)925 if res.branch is not None:926 new_head = HeadNode(branch_name = res.branch.name)927 else:928 new_head = HeadNode(commit_id = commit.commit_id)929 rit.set_head(new_head)930 if restored:931 logger.info("Successful checkout. Commit this checkout to get a clean rit status.")932 else:933 logger.info("Switched to new ref")934 return res935def checkout_orphan(*,936 root_rit_dir: str,937 name: str):938 '''939 Move head to point at branch with name name. Ensures that name doesn't already940 exist. This results in the head not pointing to any commit. Next commit will941 create a new branch.942 '''943 logger.debug('checkout_orphan')944 logger.debug(' name: %s', name)945 check_types(946 name = (name, optional_t(exact_t(str))),947 )948 validate_branch_name(name)949 rit = RitResource(root_rit_dir)950 if rit.is_branch(name):951 raise RitError("Orphan checkout failed because branch already exists.")952 new_head = HeadNode(branch_name=name)953 rit.set_head(new_head)954'''955API956You should be able to do anything you want with these functions.957'''958def query_cmd(*, root_rit_dir: str):959 '''960 Return a read only RitResource used for querying the rit directory.961 CLI users get information via stdout. That's not helpful to for python users,962 so they instead get access to the full data structure.963 For advanced users who want write access to the rit directory, construct a964 RitResource directly with prevent_mutations set to False and read the class'965 docs.966 '''967 return RitResource(root_rit_dir, prevent_mutations=True)968def init_cmd(*, root_rit_dir: str):969 logger.debug("init")970 rit = RitResource(root_rit_dir)971 rit.initialize()972def commit_cmd(*, root_rit_dir: str, msg: str):973 logger.debug('commit')974 logger.debug(' msg: %s', msg)975 check_types(976 msg = (msg, exact_t(str)),977 )978 rit = RitResource(root_rit_dir)979 commit = create_commit(rit, time.time(), msg)980 logger.info("Created commit %s: %s", commit.commit_id[:short_hash_index], commit.msg)981 return commit982def reset_cmd(*, root_rit_dir: str, ref: Optional[str], hard: bool):983 '''984 A reset tries to move the head to ref. If head is a branch, it instead moves985 the branch. If head is a commit, it just moves head to the new commit. If it's986 a hard reset, then post moving head, the working changes are removed.987 Here, a non hard reset is effectively the same as git's reset with no988 arguments. In git, a reset with no args is --mixed, and since we don't have989 staged changes, --mixed and --soft are the same in this context. For that990 reason, here, we just use the terminology reset instead of soft reset or mixed991 reset.992 '''993 logger.debug('reset')994 logger.debug(' ref: %s', ref)995 logger.debug(' hard: %s', hard)996 check_types(997 ref = (ref, optional_t(exact_t(str))),998 hard = (hard, exact_t(bool)),999 )1000 rit = RitResource(root_rit_dir)1001 res = resolve_ref(rit, ref)1002 if res.commit is None:1003 raise RitError("Unable to resolve ref to commit: %s", ref)1004 if rit.head.branch_name is not None:1005 branch = rit.get_branch(rit.head.branch_name, ensure=False)1006 if branch is None:1007 branch = Branch(rit.head.branch_name, res.commit.commit_id)1008 else:1009 branch.commit_id = res.commit.commit_id1010 rit.set_branch(branch)1011 else:1012 new_head = HeadNode(commit_id = res.commit.commit_id)1013 rit.set_head(new_head)1014 restored = False1015 if hard:1016 restored = True1017 restore_to_commit(rit, res.commit)1018 if restored:1019 logger.info("Successful reset. Commit this checkout to get a clean rit status.")1020 else:1021 logger.info("Successful reset")1022 return res1023def checkout_cmd(*, root_rit_dir: str, orphan: bool, ref_or_name: str, force: Optional[bool]):1024 '''1025 Forwards to checkout_ref or checkout_orphan1026 '''1027 logger.debug('checkout')1028 logger.debug(' orphan: %s', orphan)1029 logger.debug(' ref_or_name: %s', ref_or_name)1030 logger.debug(' force: %s', force)1031 check_types(...
test_rit.py
Source:test_rit.py
...238 assert new_commit.parent_commit_id is None239 # get rit res240 rit_res = query_cmd(**base_kwargs)241 # verify commit id matches what was returned242 new_commit_lookup = rit_res.get_commit(new_commit.commit_id, ensure=True)243 assert new_commit_lookup.commit_id is not None244 assert new_commit_lookup.commit_id == new_commit.commit_id245 assert new_commit_lookup.parent_commit_id == new_commit.parent_commit_id246 assert new_commit_lookup.create_time == new_commit.create_time247 assert new_commit_lookup.msg == new_commit.msg248 root_commit_id = new_commit_lookup.commit_id249 # verify that head is still on branch250 assert rit_res.head.branch_name == 'otest'251 assert rit_res.head.commit_id is None252 # verify that branch matches claimed commit253 new_branch = rit_res.get_branch('otest')254 assert new_branch is not None255 assert new_branch.commit_id == new_commit.commit_id256 # add files257 branch_cmd(**base_kwargs, name="otest_root", ref=None, force=False, delete=False)258 otest_a_file = touch(root_rit_dir, 'otest_a')259 commit_cmd(**base_kwargs, msg="add a")260 branch_cmd(**base_kwargs, name="otest_a", ref=None, force=False, delete=False)261 otest_b_file = touch(root_rit_dir, 'otest_b')262 top_commit = commit_cmd(**base_kwargs, msg="add b")263 branch_cmd(**base_kwargs, name="otest_b", ref=None, force=False, delete=False)264 otest_c_file = touch(root_rit_dir, 'otest_c')265 rit_res = query_cmd(**base_kwargs)266 assert rit_res.head.branch_name == 'otest'267 otest_root_branch = rit_res.get_branch('otest_root')268 assert otest_root_branch is not None269 otest_a_branch = rit_res.get_branch('otest_a')270 assert otest_a_branch is not None271 otest_b_branch = rit_res.get_branch('otest_b')272 assert otest_b_branch is not None273 otest_branch = rit_res.get_branch('otest')274 assert otest_branch is not None275 assert otest_branch.commit_id == otest_b_branch.commit_id276 assert top_commit.commit_id == otest_branch.commit_id277 assert top_commit.parent_commit_id == otest_a_branch.commit_id278 assert root_commit_id == rit_res.get_commit(otest_a_branch.commit_id, ensure=True).parent_commit_id279 commit_chain = [root_commit_id, top_commit.parent_commit_id, top_commit.commit_id]280 # make orphan and make sure changes are still there281 checkout_cmd(**base_kwargs, orphan=True, ref_or_name='otest_2', force=None)282 assert os.path.exists(otest_a_file)283 assert os.path.exists(otest_b_file)284 assert os.path.exists(otest_c_file)285 checkout_cmd(**base_kwargs, orphan=True, ref_or_name='otest_3', force=None)286 assert os.path.exists(otest_a_file)287 assert os.path.exists(otest_b_file)288 assert os.path.exists(otest_c_file)289 # get rit res290 rit_res = query_cmd(**base_kwargs)291 assert rit_res.get_branch('otest_2') is None292 assert rit_res.get_branch('otest_3') is None293 assert rit_res.head.branch_name == 'otest_3'294 reset_cmd(**base_kwargs, ref=commit_chain[0], hard=False)295 rit_res = query_cmd(**base_kwargs)296 assert rit_res.head.branch_name == 'otest_3'297 assert rit_res.head.commit_id is None298 this_branch = rit_res.get_branch('otest_3', ensure=True)299 assert this_branch is not None300 assert this_branch.commit_id == commit_chain[0]301 assert os.path.exists(otest_a_file)302 assert os.path.exists(otest_b_file)303 assert os.path.exists(otest_c_file)304 reset_cmd(**base_kwargs, ref=None, hard=False)305 rit_res = query_cmd(**base_kwargs)306 assert rit_res.get_branch('otest_2') is None307 assert rit_res.head.branch_name == 'otest_3'308 reset_cmd(**base_kwargs, ref=rit_lib.head_ref_name, hard=False)309 rit_res = query_cmd(**base_kwargs)310 assert rit_res.get_branch('otest_2') is None311 assert rit_res.head.branch_name == 'otest_3'312 reset_cmd(**base_kwargs, ref=commit_chain[1], hard=False)313 rit_res = query_cmd(**base_kwargs)314 assert rit_res.head.branch_name == 'otest_3'315 assert rit_res.head.commit_id is None316 this_branch = rit_res.get_branch('otest_3', ensure=True)317 assert this_branch is not None318 assert this_branch.commit_id == commit_chain[1]319 assert os.path.exists(otest_a_file)320 assert os.path.exists(otest_b_file)321 assert os.path.exists(otest_c_file)322 reset_cmd(**base_kwargs, ref=None, hard=False)323 rit_res = query_cmd(**base_kwargs)324 assert rit_res.head.branch_name == 'otest_3'325 reset_cmd(**base_kwargs, ref=rit_lib.head_ref_name, hard=False)326 rit_res = query_cmd(**base_kwargs)327 assert rit_res.head.branch_name == 'otest_3'328 reset_cmd(**base_kwargs, ref=commit_chain[2], hard=False)329 rit_res = query_cmd(**base_kwargs)330 assert rit_res.head.branch_name == 'otest_3'331 assert rit_res.head.commit_id is None332 this_branch = rit_res.get_branch('otest_3', ensure=True)333 assert this_branch is not None334 assert this_branch.commit_id == commit_chain[2]335 assert os.path.exists(otest_a_file)336 assert os.path.exists(otest_b_file)337 assert os.path.exists(otest_c_file)338 checkout_cmd(**base_kwargs, orphan=False, ref_or_name=commit_chain[2], force=False)339 rit_res = query_cmd(**base_kwargs)340 assert rit_res.head.branch_name is None341 assert rit_res.head.commit_id == commit_chain[2]342 reset_cmd(**base_kwargs, ref=commit_chain[0], hard=False)343 rit_res = query_cmd(**base_kwargs)344 assert rit_res.head.branch_name is None345 assert rit_res.head.commit_id == commit_chain[0]346 assert os.path.exists(otest_a_file)347 assert os.path.exists(otest_b_file)348 assert os.path.exists(otest_c_file)349 reset_cmd(**base_kwargs, ref=None, hard=False)350 rit_res = query_cmd(**base_kwargs)351 assert rit_res.head.commit_id == commit_chain[0]352 reset_cmd(**base_kwargs, ref=rit_lib.head_ref_name, hard=False)353 rit_res = query_cmd(**base_kwargs)354 assert rit_res.head.commit_id == commit_chain[0]355 reset_cmd(**base_kwargs, ref=commit_chain[2], hard=False)356 rit_res = query_cmd(**base_kwargs)357 assert rit_res.head.branch_name is None358 assert rit_res.head.commit_id == commit_chain[2]359 assert os.path.exists(otest_a_file)360 assert os.path.exists(otest_b_file)361 assert os.path.exists(otest_c_file)362 reset_cmd(**base_kwargs, ref=commit_chain[0], hard=False)363 rit_res = query_cmd(**base_kwargs)364 assert rit_res.head.branch_name is None365 assert rit_res.head.commit_id == commit_chain[0]366 assert os.path.exists(otest_a_file)367 assert os.path.exists(otest_b_file)368 assert os.path.exists(otest_c_file)369 reset_cmd(**base_kwargs, ref=commit_chain[2], hard=True)370 rit_res = query_cmd(**base_kwargs)371 assert rit_res.head.branch_name is None372 assert rit_res.head.commit_id == commit_chain[2]373 assert os.path.exists(otest_a_file)374 assert os.path.exists(otest_b_file)375 assert not os.path.exists(otest_c_file)376 reset_cmd(**base_kwargs, ref=commit_chain[1], hard=True)377 rit_res = query_cmd(**base_kwargs)378 assert rit_res.head.branch_name is None379 assert rit_res.head.commit_id == commit_chain[1]380 assert os.path.exists(otest_a_file)381 assert not os.path.exists(otest_b_file)382 assert not os.path.exists(otest_c_file)383 reset_cmd(**base_kwargs, ref=commit_chain[0], hard=True)384 rit_res = query_cmd(**base_kwargs)385 assert rit_res.head.branch_name is None386 assert rit_res.head.commit_id == commit_chain[0]387 assert not os.path.exists(otest_a_file)388 assert not os.path.exists(otest_b_file)389 assert not os.path.exists(otest_c_file)390 reset_cmd(**base_kwargs, ref=commit_chain[1], hard=True)391 rit_res = query_cmd(**base_kwargs)392 assert rit_res.head.branch_name is None393 assert rit_res.head.commit_id == commit_chain[1]394 assert os.path.exists(otest_a_file)395 assert not os.path.exists(otest_b_file)396 assert not os.path.exists(otest_c_file)397 reset_cmd(**base_kwargs, ref=commit_chain[2], hard=True)398 rit_res = query_cmd(**base_kwargs)399 assert rit_res.head.branch_name is None400 assert rit_res.head.commit_id == commit_chain[2]401 assert os.path.exists(otest_a_file)402 assert os.path.exists(otest_b_file)403 assert not os.path.exists(otest_c_file)404 checkout_cmd(**base_kwargs, orphan=False, ref_or_name='otest_3', force=True)405 reset_cmd(**base_kwargs, ref=commit_chain[2], hard=True)406 rit_res = query_cmd(**base_kwargs)407 assert rit_res.head.branch_name == 'otest_3'408 assert rit_res.head.commit_id is None409 assert os.path.exists(otest_a_file)410 assert os.path.exists(otest_b_file)411 assert not os.path.exists(otest_c_file)412 reset_cmd(**base_kwargs, ref=commit_chain[1], hard=True)413 rit_res = query_cmd(**base_kwargs)414 assert rit_res.head.branch_name == 'otest_3'415 assert rit_res.head.commit_id is None416 assert os.path.exists(otest_a_file)417 assert not os.path.exists(otest_b_file)418 assert not os.path.exists(otest_c_file)419 reset_cmd(**base_kwargs, ref=commit_chain[0], hard=True)420 rit_res = query_cmd(**base_kwargs)421 assert rit_res.head.branch_name == 'otest_3'422 assert rit_res.head.commit_id is None423 assert not os.path.exists(otest_a_file)424 assert not os.path.exists(otest_b_file)425 assert not os.path.exists(otest_c_file)426 checkout_cmd(**base_kwargs, orphan=False, ref_or_name='deviate', force=True)427 rit_res = query_cmd(**base_kwargs)428 head_branch = rit_res.get_branch(rit_res.head.branch_name)429 head_commit_id = head_branch.commit_id430 head_commit = rit_res.get_commit(head_commit_id, ensure=True)431 reset_cmd(**base_kwargs, ref=head_commit.parent_commit_id, hard=False)432 rit_res = query_cmd(**base_kwargs)433 rit_res.get_commit(head_commit_id, ensure=True)434 prune_res = prune_cmd(**base_kwargs)435 rit_res = query_cmd(**base_kwargs)436 assert rit_res.get_commit(head_commit_id) is None437 assert len(prune_res) == 1438 assert prune_res[0] == head_commit_id439 rit_res = query_cmd(**base_kwargs)440 head_commit = rit_res.get_head_commit_id()441 checkout_cmd(**base_kwargs, orphan=False, ref_or_name=head_commit, force=True)442 reset_cmd(**base_kwargs, ref='deviate', hard=False)443 rit_res = query_cmd(**base_kwargs)444 assert rit_res.head.branch_name is None...
CA4_Python_Script.py
Source:CA4_Python_Script.py
1# Name: Paul Prew2# Student Number: 103548283# Programming for Big Data4# CA 45# This is a python program that will execute to read in a sample text data file which is named 'changes_python.txt'. 6# This program works to scrub/clean the data, and differentiate the content into 422 separate objects, 7# which can be represented as rows on a spreadsheet.8# Where possible additional data was derived from the data e.g. week number was derived from the date. 9# The data was mostly extracted from the title row. 10# Additionally, for each row the number of files actioned for each revision number was calculated, and broken out by action type e.g. A, D, M, R.11# This extract provides a count of the total number of file changes, broken out by action type. 12# This also enables analysis on the number of file changed by author, and breakdown by action type.13# In the program the 'Commit' class is used to define the elements of the 'commit data' extracted from the file.14# Each of the 422 commits, are represented as instance objects of the 'Commit' class.15# Functions operate on these objects to perform analysis on the commit data, and return the result as dictionary files. 16# Dictionary files are used for the analysys outputs, as this facilitates export as CSV. 17# Visualisation of analysis was done using charts in R Studio, and Microsoft Excel.18# When the program executes, the user is asked to confirm the action, to begin data file read.19# If successful, a message printed to the console will acknowledge this, and also display the total number of commits in the extract 20# and the total number of lines in the file.21# The user is then asked to confirm next action, to perform analysis on the commit data. 22# The program will then execute the analysis and export the results to CSV files in the working directory.23# When above step is completed, a message printed to the console will confirm this.24# The user will now be asked if they wish to print the analysis results to the console. If the user selects this option25# the results are printed on screen.26sep = 72*'-'27import datetime28import csv29import os30# Below defines the class 'Commit' which is used to define the elements of the 'commit data' extracted from the file.31class Commit:32 'class for commits'33 34 def __init__(self, revision = None, author = None, fulldatetime = None, fulldate = None, dateinmonth = None, dayinweek = None, weeknumber = None, month = None, year = None,35 comment_line_count = None, number_changes = None, number_a = None, number_d = None, number_m = None, number_r = None, changes = None, comment = None):36 self.revision = revision37 self.author = author38 self.fulldatetime = fulldatetime39 self.fulldate = fulldate40 self.dateinmonth = dateinmonth41 self.dayinweek = dayinweek42 self.weeknumber = weeknumber43 self.month = month44 self.year = year45 self.comment_line_count = comment_line_count46 self.number_changes = number_changes47 self.changes = changes48 self.number_a = number_a49 self.number_d = number_d50 self.number_m = number_m51 self.number_r = number_r52 self.comment = comment53 54# Below is a generic function that will accept as parameter an attribute from the commit data55# This allows this function to be called multiple times with different parameter values 56# e.g. revisions by author, revisions by day.57def get_revisions_by_attribute(data, attribute):58 result = {}59 for commit in data:60 value = getattr(commit, attribute)61 item = value62 result[item] = result.get(item, 0) + 163 return result64# Below function will get the total number of file changes for all 422 commits,65# and break them out by file change action type i.e. A, D, M, R,66# and also by author 67def get_file_changes_by_action(data):68 all_authors = []69 for commit in data:70 new_author = {}71 author = commit.author72 match_found = False 73 if len(all_authors) > 0:74 for record in all_authors:75 if record['Author'] == author :76 record['Action Type A'] += commit.number_a77 record['Action Type D'] += commit.number_d78 record['Action Type M'] += commit.number_m79 record['Action Type R'] += commit.number_r80 record['Total Files Actioned'] += commit.number_changes81 match_found = True82 break83 84 if len(all_authors) == 0 or match_found == False : 85 new_author = {'Author': author, 'Action Type A': commit.number_a, 'Action Type D': commit.number_d, 'Action Type M': commit.number_m, 86 'Action Type R': commit.number_r, 'Total Files Actioned' : commit.number_changes } 87 all_authors.append(new_author)88 89 return all_authors90 91def read_file(changes_file):92 data = [line.strip() for line in open(changes_file, 'r')]93 return data 94def get_commits(data):95 commits = []96 current_commit = None97 index = 0 98 author = {}99 while True:100 try:101 current_commit = Commit()102 details = data[index + 1].split('|')103 current_commit.revision = int(details[0].strip().strip('r'))104 current_commit.author = details[1].strip()105 current_commit.fulldatetime = details[2].strip() 106 current_commit.dateinmonth = int(details[2][9:11])107 current_commit.month = int(details[2][6:8])108 current_commit.year = int(details[2][0:5])109 current_commit.fulldate = datetime.date(current_commit.year, current_commit.month, current_commit.dateinmonth)110 current_commit.weeknumber = current_commit.fulldate.strftime("%W")111 current_commit.dayinweek = details[2][28:31]112 current_commit.comment_line_count = int(details[3].strip().split(' ')[0])113 current_commit.changes = data[index+2:data.index('',index+1)]114 current_commit.number_changes = len(current_commit.changes)-1 115 current_commit.number_a = 0116 current_commit.number_d = 0117 current_commit.number_m = 0118 current_commit.number_r = 0119 for i in range(current_commit.number_changes) :120 if current_commit.changes[i + 1].startswith("A /") :121 current_commit.number_a = current_commit.number_a + 1122 elif current_commit.changes[i + 1].startswith("D /") :123 current_commit.number_d = current_commit.number_d + 1124 elif current_commit.changes[i + 1].startswith("M /") :125 current_commit.number_m = current_commit.number_m + 1126 elif current_commit.changes[i + 1].startswith("R /") :127 current_commit.number_r = current_commit.number_r + 1128 index = data.index(sep, index + 1)129 current_commit.comment = data[index-current_commit.comment_line_count:index]130 commits.append(current_commit)131 except IndexError:132 break133 return commits134# this function saves the commit list (class instances) to a CSV log file135def save_commit_list_as_csv (location): 136 with open(location,"wb") as data_export:137 fnameWriter = csv.writer(data_export)138 fnameWriter.writerow(["Revision", "Author", "Full Date/Time", "Full Date", "Date in Month", "Day in Week", "Week Number", "Month", "Year", "Comment Line Count", "Number Changes", "Number A", 139 "Number D", "Number M", "Number R", "Comment"])140 for commit in results:141 fnameWriter.writerow([commit.revision, commit.author, commit.fulldatetime, commit.fulldate, commit.dateinmonth, commit.dayinweek, commit.weeknumber, commit.month,142 commit.year, commit.comment_line_count, commit.number_changes, commit.number_a, commit.number_d, commit.number_m, commit.number_r,143 commit.comment])144 data_export.close()145# this function saves the analysis results to the console, and is used where the result to save, is formatted as a single dictionary146# the function accepts as parameters: data, and location, so can be used many times 147def save_dict_as_csv (data,location):148 with open(location,'wb') as output_file:149 w = csv.writer(output_file)150 w.writerows(data.items())151 output_file.close()152# this function saves the analysis results to a CSV log file, and is used where the result to save, is formatted as a list of dictionaries153def save_dict_list_as_csv(data, location) : 154 keys = data[0].keys()155 with open(location, 'wb') as output_file:156 dict_writer = csv.DictWriter(output_file, keys)157 dict_writer.writeheader()158 dict_writer.writerows(data) 159 output_file.close()160# this function prints the analysis results to the console, and is used where the result to print, is formatted as a single dictionary161# the function accepts as parameters: data, and location, so can be used many times 162def print_dict_to_console(data,title):163 print ''164 print title165 print "="*len(title)166 ls = list()167 for key, val in data.items():168 ls.append((val,key))169 ls.sort(reverse = True)170 for key, val in ls:171 print key, val172 return ls173# this function prints the analysis results to the console, and is used where the result to print, is formatted as a list of dictionaries174def print_dict_list_to_console(data, title): 175 print ''176 print title177 print "=" * len(title)178 for row in data:179 print row180def export_analysis_data_to_CSV():181 filetoCSV = authors182 save_dict_as_csv(filetoCSV,"Python_Export_Counts of Revisions By Author.csv") 183 filetoCSV = days184 save_dict_as_csv(filetoCSV,"Python Export_Counts of Revisions By Day in Week.csv") 185 filetoCSV = change_files186 save_dict_list_as_csv(filetoCSV,"Python Export_Counts of Files Changed By Action Type.csv") 187 188# main program 189if __name__ == '__main__': 190 191 os.system('cls')192 193 print '\n***************************************\n'194 print "Analysis for sample text file"195 print "Text file is 'changes_python.txt'"196 print '\n***************************************\n'197 while True:198 try: 199 s_inp = raw_input("Enter any key to read data file and get commits: ")200 data = read_file('changes_python.txt')201 results = get_commits(data)202 # following line saves the commit list with 422 rows to a csv which acts as a log file203 save_commit_list_as_csv("Python Export_Commit List.csv") 204 print "\nFile read was successful, and data has been extracted"205 print "{} rows of commits extracted from {} lines" .format(len(results), len(data))206 print "Results saved to csv log file in working directory"207 break208 209 except:210 print "\nError while reading file. Please try again."211 print "Ensure text file is saved as 'changes_python.txt and is saved in the working directory."212 continue 213 214 s_inp = raw_input("\nEnter any key to perform analysis on commit data: ")215 216 authors = get_revisions_by_attribute(results, "author")217 days = get_revisions_by_attribute(results, "dayinweek")218 change_files = get_file_changes_by_action(results)219 220 export_analysis_data_to_CSV()221 print "\nAnalysis on commit data is completed"222 print "Results saved as CSV files in working directory"223 224 print "\nEnter 'Y' to print analysis summary to console"225 print "Enter any other key to exit program"226 choice = raw_input()227 if choice.lower() == 'y' :228 print_dict_to_console(authors, "Revisions By Author")229 print_dict_to_console(days, "Revisions By Day in Week")230 print_dict_list_to_console(change_files, "Revisions By File Changes By Author")231 else :232 print "\nExited program"233 234 235 236 ...
compare_footprint
Source:compare_footprint
...66 parser.add_argument('-c', '--commit', default=None,67 help="Commit ID to use compare footprint against base. "68 "Default is HEAD or working tree.")69 return parser.parse_args()70def get_git_commit(commit):71 commit_id = None72 proc = subprocess.Popen('git rev-parse %s' % commit, stdout=subprocess.PIPE,73 cwd=os.environ.get('ZEPHYR_BASE'), shell=True)74 if proc.wait() == 0:75 commit_id = proc.stdout.read().decode("utf-8").strip()76 return commit_id77def sanity_results_filename(commit=None, cwd=os.environ.get('ZEPHYR_BASE')):78 if not commit:79 file_name = "tmp.csv"80 else:81 if commit == RELEASE_DATA:82 file_name = RELEASE_DATA83 else:84 file_name = "%s.csv" % commit85 return os.path.join(cwd,'scripts', 'sanity_chk', file_name)86def git_checkout(commit, cwd=os.environ.get('ZEPHYR_BASE')):87 proc = subprocess.Popen('git diff --quiet', stdout=subprocess.PIPE,88 stderr=subprocess.STDOUT, cwd=cwd, shell=True)89 if proc.wait() != 0:90 raise Exception("Cannot continue, you have unstaged changes in your working tree")91 proc = subprocess.Popen('git reset %s --hard' % commit,92 stdout=subprocess.PIPE,93 stderr=subprocess.STDOUT,94 cwd=cwd, shell=True)95 if proc.wait() == 0:96 return True97 else:98 logger.error(proc.stdout.read())99 return False100def run_sanity_footprint(commit=None, cwd=os.environ.get('ZEPHYR_BASE'),101 output_file=None):102 if not output_file:103 output_file = sanity_results_filename(commit)104 cmd = '/bin/bash -c "source ./zephyr-env.sh && twister'105 cmd += ' +scripts/sanity_chk/sanity_compare.args -o %s"' % output_file106 logger.debug('Sanity (%s) %s' %(commit, cmd))107 proc = subprocess.Popen(cmd, stdout=subprocess.PIPE,108 cwd=cwd, shell=True)109 output,_ = proc.communicate()110 if proc.wait() == 0:111 logger.debug(output)112 return True113 logger.error("Couldn't build footprint apps in commit %s" % commit)114 logger.error(output)115 raise Exception("Couldn't build footprint apps in commit %s" % commit)116def run_footprint_build(commit=None):117 logging.debug("footprint build for %s" % commit)118 if not commit:119 run_sanity_footprint()120 else:121 cmd = "git clone --no-hardlinks %s" % os.environ.get('ZEPHYR_BASE')122 tmp_location = os.path.join(tempfile.gettempdir(),123 os.path.basename(os.environ.get('ZEPHYR_BASE')))124 if os.path.exists(tmp_location):125 shutil.rmtree(tmp_location)126 logging.debug("clonning into %s" % tmp_location)127 proc = subprocess.Popen(cmd, stdout=subprocess.PIPE,128 stderr=subprocess.STDOUT,129 cwd=tempfile.gettempdir(), shell=True)130 if proc.wait() == 0:131 if git_checkout(commit, tmp_location):132 run_sanity_footprint(commit, tmp_location)133 else:134 logger.error(proc.stdout.read())135 shutil.rmtree(tmp_location, ignore_errors=True)136 return True137def read_sanity_report(filename):138 data = []139 with open(filename) as fp:140 tmp = csv.DictReader(fp)141 for row in tmp:142 data.append(row)143 return data144def get_footprint_results(commit=None):145 sanity_file = sanity_results_filename(commit)146 if (not os.path.exists(sanity_file) or not commit) and commit != RELEASE_DATA:147 run_footprint_build(commit)148 return read_sanity_report(sanity_file)149def tree_changes():150 proc = subprocess.Popen('git diff --quiet', stdout=subprocess.PIPE,151 cwd=os.environ.get('ZEPHYR_BASE'), shell=True)152 if proc.wait() != 0:153 return True154 return False155def get_default_current_commit():156 if tree_changes():157 return None158 else:159 return get_git_commit('HEAD')160def get_default_base_commit(current_commit):161 if not current_commit:162 if tree_changes():163 return get_git_commit('HEAD')164 else:165 return get_git_commit('HEAD~1')166 else:167 return get_git_commit('%s~1'%current_commit)168def build_history(b_commit=None, c_commit=None):169 if not GIT_ENABLED:170 logger.info('Working on current tree, not git enabled.')171 current_commit = None172 base_commit = RELEASE_DATA173 else:174 if not c_commit:175 current_commit = get_default_current_commit()176 else:177 current_commit = get_git_commit(c_commit)178 if not b_commit:179 base_commit = get_default_base_commit(current_commit)180 else:181 base_commit = get_git_commit(b_commit)182 if not base_commit:183 logger.error("Cannot resolve base commit")184 return185 logger.info("Base: %s" % base_commit)186 logger.info("Current: %s" % (current_commit if current_commit else187 'working space'))188 current_results = get_footprint_results(current_commit)189 base_results = get_footprint_results(base_commit)190 deltas = compare_results(base_results, current_results)191 print_deltas(deltas)192def compare_results(base_results, current_results):193 interesting_metrics = [("ram_size", int),194 ("rom_size", int)]195 results = {}...
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 localstack 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?
