How to use oneup method in pyatom

Best Python code snippet using pyatom_python

crawl.py

Source:crawl.py Github

copy

Full Screen

1import yaml2from pathlib import Path3import os4import git5import time6import shutil7import subprocess8from time import gmtime, strftime9## anthos-service-mesh-samples 10# # if no metadata name, then format: <product-prefix>_<oneup>_<filename-without-extention>_<kind>11## anthos config management12# 13google_license = """14# Copyright 2022 Google LLC15#16# Licensed under the Apache License, Version 2.0 (the "License");17# you may not use this file except in compliance with the License.18# You may obtain a copy of the License at19#20# http://www.apache.org/licenses/LICENSE-2.021#22# Unless required by applicable law or agreed to in writing, software23# distributed under the License is distributed on an "AS IS" BASIS,24# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.25# See the License for the specific language governing permissions and26# limitations under the License.27"""28google_license_set = set(google_license.splitlines())29all_results = {}30repo = None31def remove_tags(fn):32 count_remove_tags = 033 ## this will remove ALL the tags in the yaml34 with open(fn,"r") as f:35 lines = f.readlines()36 with open(fn,"w") as f:37 for line in lines:38 if not line.startswith("# [") :39 f.write(line)40 continue41 else:42 count_remove_tags += 143 # if not line.startswith("# [END"):44 # f.write(line)45 print("REMOVED THE TAGS at " +fn)46 print("REMOVED "+ count_remove_tags//2 + "TAGS")47 return48def generate_region_tag(product, twoup, oneup, fn, snippet):49 # get the specific Kubernetes resource type, if listed. 50 if 'kind' not in snippet:51 resource_type = "yaml"52 else:53 resource_type = snippet['kind'].lower()54 filename = os.path.basename(fn)55 print(twoup, oneup)56 # have the snippet name match the k8s resource name - otherwise, match to filename 57 if 'metadata' not in snippet:58 if filename.endswith('.yaml'):59 # if no metadata name, then format: <product-prefix>_<oneup>_<filename-without-extention>_<kind>60 filename = filename[:-5]61 tag = "{}_{}_{}_{}_{}".format(product, twoup,oneup, filename ,resource_type)62 ## if the file is a shell 63 elif filename.endswith('.yml'):64 filename = filename[:-4]65 tag = "{}_{}_{}_{}_{}".format(product, twoup,oneup, filename ,resource_type)66 else: 67 if filename.endswith('.yaml'):68 filename = filename[:-5]69 metadataName = snippet['metadata']['name'].lower()70 tag = "{}_{}_{}_{}_{}".format(product, oneup, filename, resource_type,metadataName)71 elif filename.endswith('.yml'):72 filename = filename[:-4]73 metadataName = snippet['metadata']['name'].lower()74 tag = "{}_{}_{}_{}_{}".format(product, oneup, filename ,resource_type,metadataName)75 tag = tag.replace("-", "_").replace(".","_")76 return tag77def generate_shell_tag(product, twoup,fn):78 filename = os.path.basename(fn)79 filename = filename[:-3]80 tag = "{}_{}_{}".format(product, twoup, filename )81 tag = tag.replace("-", "_")82 return tag83def process_file(product, twoup, oneup, fn):84 if (oneup=="templates") or (oneup==".github") or (twoup==".github") or ((oneup=="kubernetes-manifests") and (twoup=="bank-of-anthos")) or ((oneup=="istio-manifests") and (twoup=="bank-of-anthos")):85 return 86 global all_results87 # check if there are an existing region tag, then don't touch them 88 with open(fn) as file:89 if '# [START' in file.read():90 return91 file.close()92 # with open(fn, 'r').read().find('# [START '):93 yaml_comments = {}94 with open(fn) as file:95 # do not process helm charts96 results = {}97 original_contents = file.readlines()98 # print(original_contents[:14], google_license.splitlines())99 # stripped = [s.strip() for s in original_contents[:14]]100 # print(stripped==google_license.splitlines())101 if len(original_contents)<14 or original_contents[0].find('apiVersion')>=0:102 start = 0103 end = 0104 else:105 # start and end of the license if present106 start = 0107 end = 12108 # print(original_contents[end].strip()!= "# limitations under the License.")109 if original_contents[end].strip()!= "# limitations under the License.":110 start = 1111 end = 13112 ## this is so janky, but the idea is to find the comments (not including license), save comment and line location, add the length of the google license and hope it works113 yaml_sections_count = 0114 for i,line in enumerate(original_contents[end+1:]):115 check_line = line.lstrip() # remove whitespace from left of the string116 if len(check_line)>1 and check_line[0]=='#':117 # print(i, end, yaml_sections_count, line)118 if end>0:119 yaml_comments[i + 15 + (yaml_sections_count*3)] = line120 else:121 yaml_comments[27 + 15 + (yaml_sections_count*3)] = line122 if line.find("kind:")>=0 and end == 0:123 yaml_sections_count+=1124 continue125 if line.find("kind:")>=0 and i>17: # skip over the first one126 # print(line.find("kind:"))127 yaml_sections_count += 1128 129 # print(yaml_comments)130 file.seek(0)131 documents = yaml.load_all(file, Loader=yaml.FullLoader)132 for snippet in documents:133 if snippet is None:134 continue135 tag = generate_region_tag(product, twoup, oneup, fn, snippet)136 # handle duplicates.137 if tag in all_results:138 print("⭐️ tag {} already in all_results, adding a #". format(tag))139 x = 2140 numbered_tag = tag + str(x)141 while numbered_tag in all_results:142 print("tag {} already in all_results, adding a #". format(numbered_tag))143 x = x + 1144 numbered_tag = tag + str(x)145 print("adding numbered tag: {} to all_results \n".format(numbered_tag))146 results[numbered_tag] = snippet147 all_results[numbered_tag] = snippet148 else:149 results[tag] = snippet150 all_results[tag] = snippet151 file.close()152 # write new YAML file with google license, START and END153 with open(fn, 'w+') as output:154 output.write(google_license + "\n")155 for tag, snippet in results.items():156 start = "# [START {}]".format(tag)157 end = "# [END {}]".format(tag)158 output.write(start + "\n")159 yaml.dump(snippet, output, sort_keys=False)160 output.write(end + "\n")161 output.write("---\n")162 output.close()163 ## looping through the new file, and adding comments back in 164 # safely read then write to the file165 with open(fn,'r') as no_comments:166 buf = no_comments.readlines()167 no_comments.close() 168 for i, line in yaml_comments.items():169 buf.insert(i,line)170 with open(fn,'w') as outfile:171 buf = "".join(buf)172 outfile.write(buf)173 outfile.close()174def process_file_shell(product, oneup, twoup, fn):175 if (oneup=="templates") or (oneup==".github") or (twoup==".github"):176 return 177 global all_results178 # check if there are an existing region tag, then don't touch them 179 with open(fn) as file:180 if '# [START' in file.read():181 return182 file.close()183 tag = generate_shell_tag(product,oneup,fn)184 start = "# [START {}]".format(tag)185 end = "# [END {}]".format(tag)186 insert_start = False187 with open(fn) as f:188 filIn = f.readlines()189 for i,line in enumerate(filIn):190 if line == "# limitations under the License.":191 insert_start = True192 break193 if insert_start:194 filIn.insert(i, start+ "\n")195 f.close()196 197 if insert_start == False:198 # if the file has no license, add license, add tags, add existing content, add end tag199 with open(fn) as file:200 contents = file.read()201 with open(fn,'w') as file:202 file.write(google_license + "\n")203 file.write(start + "\n")204 file.write(contents + "\n")205 file.write(end)206 file.close()207 else:208 with open(fn, 'a') as file:209 file.write(end)210 file.close()211def clone_repo(id_rsa, known_hosts, github_repository, branch, local_path):212 global repo213 # prep to clone214 my_env = os.environ.copy()215 my_env["SSH_PRIVATE_KEY"] = id_rsa216 my_env["KNOWN_HOSTS"] = known_hosts217 dir_path = os.path.dirname(os.path.realpath(__file__))218 cmd = '{}/clone_prep.sh'.format(dir_path)219 a = subprocess.run(cmd, stdout=subprocess.PIPE, env=my_env)220 print(a.stdout.decode('utf-8'))221 # clone222 repo_clone_url = 'git@github.com:{}.git'.format(github_repository)223 repo = git.Repo.clone_from(repo_clone_url, local_path)224 repo.git.checkout(branch)225def push_to_repo(local_path, branch):226 global repo227 dt = strftime("%m/%d/%Y %h:%M:%S", gmtime())228 dir_path = os.path.dirname(os.path.realpath(__file__))229 my_env = os.environ.copy()230 my_env["LOCAL_PATH"] = local_path231 my_env["BRANCH"] = branch232 my_env["COMMIT_MESSAGE"] = '[bot] generate YAML region tags {}'.format(dt)233 cmd = '{}/push.sh'.format(dir_path)234 # use script to push to github, as yamlbot235 # (no error catching / logs when doing this in python)236 a = subprocess.run(cmd, stdout=subprocess.PIPE, env=my_env)237 print(a.stdout.decode('utf-8'))238def get_repo():239 ## get reference to the repo, and create a local_branch named yaml_tags240 input_link = input('Please input the github repo github link if you do not have it cloned, or the path if its local:')241 # clone the repo if it doesn't exist already242 if input_link[:5]=="https":243 repo_directory = "/"244 repo = git.clone(input_link)245 repo.git.checkout('test')246 else: ## else refer to a local one247 repo = git.Repo(input_link)248 repo_directory = input_link249 yaml_branch = repo.create_head("drift_region_tags")250 repo.head.reference = yaml_branch251 print(repo.head.reference)252 return repo, repo_directory253def log_results():254 global all_results255 print("✅ success: total resources processed: {}".format(len(all_results.keys())))256if __name__ == "__main__":257 env_file = input("0: env file, or 1:existing clone of the repo local, or 'REMOVE_TAGS' to remove all the tags from sh and yaml: ")258 if env_file=='0':259 id_rsa = os.environ['ID_RSA']260 if id_rsa == "":261 print("Error: ID_RSA env variable must be set")262 exit(1)263 known_hosts = os.environ['KNOWN_HOSTS']264 if known_hosts == "":265 print("Error: KNOWN_HOSTS env variable must be set")266 exit(1)267 product = os.environ['PRODUCT']268 if product == "":269 print("Error: PRODUCT env variable must be set")270 exit(1)271 github_repo_name = os.environ['GITHUB_REPOSITORY']272 if github_repo_name == "":273 print("Error: GITHUB_REPOSITORY env variable must be set.")274 exit(1)275 branch = os.environ['GITHUB_REF']276 if branch == "":277 print("Error: GITHUB_REF env variable must be set.")278 exit(1)279 # clone repo280 local_path = "/tmp/{}".format(github_repo_name)281 shutil.rmtree(local_path, ignore_errors=True)282 clone_repo(id_rsa, known_hosts, github_repo_name, branch, local_path)283 # prepare to process snippets284 path = Path(local_path)285 for p in path.rglob("*.yaml"):286 filename = p.name287 fullparent = str(p.parent)288 fn = fullparent + "/" + filename289 print("processing: {}".format(fn))290 spl = fullparent.split("/")291 oneup = spl[-1]292 twoup = spl[-2]293 process_file(product, twoup, oneup, fn)294 push_to_repo(local_path, branch)295 log_results()296 elif env_file == '1':297 directory, local_path = get_repo()298 # prod_prefix = input("Input the product prefix:")299 prod_prefix = 'anthosconfig'300 path = Path(local_path)301 print("PROCESSING YAML")302 for p in path.rglob("*.yaml"):303 tempFile = open(p)304 if 'kind:' in tempFile.read():305 filename = p.name306 fullparent = str(p.parent)307 fn = fullparent + "/" + filename308 print("processing: {}".format(fn))309 spl = fullparent.split("/")310 oneup = spl[-1]311 twoup = spl[-2]312 if twoup=="repos" or oneup == "repos": ## if the yaml file is in the root of the repo313 twoup = ""314 if filename=="skaffold":315 continue316 if oneup=="release-cluster":317 continue318 process_file(prod_prefix, twoup, oneup, fn)319 320 # for p in path.rglob("*.yml"):321 # filename = p.name322 # fullparent = str(p.parent)323 # fn = fullparent + "/" + filename324 # print("processing: {}".format(fn))325 # spl = fullparent.split("/")326 # oneup = spl[-1]327 # twoup = spl[-2]328 # process_file(prod_prefix, twoup, oneup, fn)329 # capturing shell scripts to tag330 print("PROCESSING SHELL")331 for p in path.rglob("*.sh"):332 filename = p.name333 fullparent = str(p.parent)334 fn = fullparent + "/" + filename335 print("processing: {}".format(fn))336 spl = fullparent.split("/")337 oneup = spl[-1]338 twoup = spl[-2]339 # print("===================",twoup,oneup)340 process_file_shell(prod_prefix, oneup, twoup, fn)341 # push_to_repo(local_path, branch)342 log_results()343 elif env_file =="REMOVE_TAGS":344 print("************** REMOVING TAGS")345 directory, local_path = get_repo()346 path = Path(local_path)347 for p in path.rglob("*.yaml"):348 filename = p.name349 fullparent = str(p.parent)350 fn = fullparent + "/" + filename351 remove_tags(fn)352 for p in path.rglob("*.yml"):353 filename = p.name354 fullparent = str(p.parent)355 fn = fullparent + "/" + filename...

Full Screen

Full Screen

views.py

Source:views.py Github

copy

Full Screen

1from django.shortcuts import render, redirect2from django.http import HttpResponse3from .models import Outsourcing, Manufacture, As, Logistics4from django.contrib.auth.decorators import login_required, permission_required5from django.contrib import messages6from django.utils import timezone7from .forms import AsForm8from django.core.paginator import Paginator9from django.db.models import Q10from oneup.resources import OutResource, ManResource, LogResource, AsResource11@login_required(login_url='common:login')12@permission_required('oneup.view_outsourcing', login_url='oneup:deny', raise_exception=False)13def index(request):14 """15 out 목록 출력16 """17 # 입력 파라미터18 page = request.GET.get('page', '1') # 페이지19 kw = request.GET.get('kw', '') # 검색어20 so = request.GET.get('so', 'recent') # 정렬기준21 # 정렬22 if so == 'code':23 outsourcing_list = Outsourcing.objects.order_by('ossn', '-osdate')24 elif so == 'que':25 outsourcing_list = Outsourcing.objects.order_by('-osqu', '-osdate')26 else: # recent27 outsourcing_list = Outsourcing.objects.order_by('-osdate')28 # 조회29 if kw:30 outsourcing_list = outsourcing_list.filter(31 Q(ossn__icontains=kw) | # 부품코드 검색32 Q(osqu__icontains=kw) | # 납품수량 검색33 Q(osdate__icontains=kw) # 납품일시 검색34 ).distinct()35 # 페이징처리36 paginator = Paginator(outsourcing_list, 10) # 페이지당 10개씩 보여주기37 page_obj = paginator.get_page(page)38 context = {'outsourcing_list': page_obj, 'page': page, 'kw': kw, 'so': so} # page, kw, so가 추가되었다.39 return render(request, 'oneup/outsourcing_list.html', context)40@login_required(login_url='common:login')41@permission_required('oneup.view_manufacture', login_url='oneup:deny', raise_exception=False)42def manufacturer(request):43 """44 manu 목록 출력45 """46 # 입력 파라미터47 page = request.GET.get('page', '1') # 페이지48 kw = request.GET.get('kw', '') # 검색어49 so = request.GET.get('so', 'recent') # 정렬기준50 # 정렬51 if so == 'pro':52 manufacture_list = Manufacture.objects.order_by('-proname', '-mfdate')53 elif so == 'mfsn':54 manufacture_list = Manufacture.objects.order_by('-mfsn', '-mfdate')55 elif so == 'ori':56 manufacture_list = Manufacture.objects.order_by('-origin', '-mfdate')57 elif so == 'half':58 manufacture_list = Manufacture.objects.order_by('-halfpro', '-mfdate')59 elif so == 'work':60 manufacture_list = Manufacture.objects.order_by('-workercode', '-mfdate')61 elif so == 'mfst':62 manufacture_list = Manufacture.objects.order_by('-mfsta', '-mfdate')63 else: # recent64 manufacture_list = Manufacture.objects.order_by('-mfdate')65 # 조회66 if kw:67 manufacture_list = manufacture_list.filter(68 Q(proname__icontains=kw) | # 제품명 검색69 Q(mfsn__icontains=kw) | # 시리얼넘버 검색70 Q(origin__icontains=kw) | # 원자재 검색71 Q(halfpro__icontains=kw) | # 반제품 검색72 Q(workercode__icontains=kw) | # 작업자코드 검색73 Q(mfsta__icontains=kw) | # 제작상태 검색74 Q(mfdate__icontains=kw) # 업데이트 일시 검색75 ).distinct()76 # 페이징처리77 paginator = Paginator(manufacture_list, 10) # 페이지당 10개씩 보여주기78 page_obj = paginator.get_page(page)79 context = {'manufacture_list': page_obj, 'page': page, 'kw': kw, 'so': so} # page, kw, so가 추가되었다.80 return render(request, 'oneup/manufacture_list.html', context)81@login_required(login_url='common:login')82@permission_required('oneup.view_logistics', login_url='oneup:deny', raise_exception=False)83def logistics(request):84 """85 log 목록 출력86 """87 # 입력 파라미터88 page = request.GET.get('page', '1') # 페이지89 kw = request.GET.get('kw', '') # 검색어90 so = request.GET.get('so', 'logn') # 정렬기준91 # 정렬92 if so == 'logcu':93 logistics_list = Logistics.objects.order_by('-logcustom')94 elif so == 'logsn':95 logistics_list = Logistics.objects.order_by('-logsn')96 elif so == 'wh':97 logistics_list = Logistics.objects.order_by('-wh')98 elif so == 'dis':99 logistics_list = Logistics.objects.order_by('-dis')100 elif so == 'agen':101 logistics_list = Logistics.objects.order_by('-agen')102 else: # logn103 logistics_list = Logistics.objects.order_by('-logname')104 # 조회105 if kw:106 logistics_list = logistics_list.filter(107 Q(logcustom__icontains=kw) | # 고객 검색108 Q(logsn__icontains=kw) | # 시리얼넘버 검색109 Q(wh__icontains=kw) | # 보관창고 검색110 Q(dis__icontains=kw) | # 총판 검색111 Q(agen__icontains=kw) | # 대리점 검색112 Q(logname__icontains=kw) # 제품명 검색113 ).distinct()114 # 페이징처리115 paginator = Paginator(logistics_list, 10) # 페이지당 10개씩 보여주기116 page_obj = paginator.get_page(page)117 context = {'logistics_list': page_obj, 'page': page, 'kw': kw, 'so': so} # page, kw, so가 추가되었다.118 return render(request, 'oneup/logistics_list.html', context)119@login_required(login_url='common:login')120@permission_required('oneup.view_as', login_url='oneup:deny', raise_exception=False)121def after(request):122 """123 as 목록 출력124 """125 # 입력 파라미터126 page = request.GET.get('page', '1') # 페이지127 kw = request.GET.get('kw', '') # 검색어128 so = request.GET.get('so', 'recent') # 정렬기준129 # 정렬130 if so == 'asname':131 as_list = As.objects.order_by('asname', '-asdate')132 elif so == 'pronu':133 as_list = As.objects.order_by('pronumber', '-asdate')134 elif so == 'assta':135 as_list = As.objects.order_by('assta', '-asdate')136 else: # recent137 as_list = As.objects.order_by('-asdate')138 # 조회139 if kw:140 as_list = as_list.filter(141 Q(asname__icontains=kw) | # 고객명 검색142 Q(ph__icontains=kw) | # 전화번호 검색143 Q(pronumber__icontains=kw) | # 제품번호 검색144 Q(assta__icontains=kw) | # A/S현황 검색145 Q(record__icontains=kw) | # 통화녹음 검색146 Q(asdate__icontains=kw) # 등록일시 검색147 ).distinct()148 # 페이징처리149 paginator = Paginator(as_list, 10) # 페이지당 10개씩 보여주기150 page_obj = paginator.get_page(page)151 context = {'as_list': page_obj, 'page': page, 'kw': kw, 'so': so} # page, kw, so가 추가되었다.152 return render(request, 'oneup/as_list.html', context)153@login_required(login_url='common:login')154def main(request):155 """156 메인 페이지 : 이동할 DB 선택157 """158 return render(request, 'oneup/main.html')159def deny(request):160 """161 권한없음162 """163 return render(request, 'oneup/deny.html')164def as_create(request):165 """166 as 정보입력167 """168 if request.method == 'POST':169 form = AsForm(request.POST)170 if form.is_valid():171 As = form.save(commit=False)172 As.asdate = timezone.now()173 As.save()174 messages.success(request, '제출해 주셔서 감사합니다!')175 return redirect('oneup:as_create')176 else:177 form = AsForm()178 context = {'form': form}179 return render(request, 'oneup/as_form.html', context)180def export1(request):181 out_resource = OutResource()182 dataset = out_resource.export()183 response = HttpResponse(dataset.xls, content_type='text/xls')184 response['Content-Disposition'] = 'attachment; filename="outsourcing.xls"'185 return response186def export2(request):187 man_resource = ManResource()188 dataset = man_resource.export()189 response = HttpResponse(dataset.xls, content_type='text/xls')190 response['Content-Disposition'] = 'attachment; filename="manufacture.xls"'191 return response192def export3(request):193 log_resource = LogResource()194 dataset = log_resource.export()195 response = HttpResponse(dataset.xls, content_type='text/xls')196 response['Content-Disposition'] = 'attachment; filename="logistics.xls"'197 return response198def export4(request):199 as_resource = AsResource()200 dataset = as_resource.export()201 response = HttpResponse(dataset.xls, content_type='text/xls')202 response['Content-Disposition'] = 'attachment; filename="AS.xls"'...

Full Screen

Full Screen

test_sprites.py

Source:test_sprites.py Github

copy

Full Screen

...3@pytest.fixture4def mario():5 return sprites.Mario(1,2)6@pytest.fixture7def oneup():8 return sprites.OneUp(1,2)9@pytest.fixture10def fireflower():11 return sprites.FireFlower(1,2)12@pytest.fixture13def mushroom():14 return sprites.Mushroom(1,2)15@pytest.fixture16def star():17 return sprites.Star(1,2)18@pytest.fixture19def brick():20 return sprites.Brick(1,0)21@pytest.fixture22def pipe():23 return sprites.Pipe(1,2)24def test_mario(mario):25 """ test if mario has the correct instance attributes and method works correctly """26 assert hasattr(mario, 'image')27 assert hasattr(mario, 'rect')28 assert hasattr(mario, 'rect.x')29 assert hasattr(mario, 'rect.y')30 assert mario.rect.x == 231 assert mario.rect.y == 132 mario.update(2,1)33 assert mario.rect.x == 134 assert mario.rect.y == 235def test_oneup(oneup):36 """ test if oneup has the correct instance attributes and method works correctly """37 assert hasattr(oneup, '_image')38 assert hasattr(oneup, 'image')39 assert hasattr(oneup, 'rect')40 assert hasattr(oneup, 'rect.x')41 assert hasattr(oneup, 'rect.y')42 assert oneup.rect.x == 243 assert oneup.rect.y == 144 45 oneup.update(3,4)46 assert oneup.rect.x == 447 assert oneup.rect.y == 348def test_fireflower(fireflower):49 """ test if fireflower has the correct instance attributes and method works correctly """...

Full Screen

Full Screen

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:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run pyatom 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!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful