Best Python code snippet using Kiwi_python
ctt.py
Source:ctt.py
...76 parser.add_argument('--attach', action='store', dest='attachment', nargs=2, required=True) 77 args = parser.parse_args()78 if args.attachment[0] and args.attachment[1]: 79 create_attachment(args.attachment[0],args.attachment[1],attach_location,date,updatedby)80 update_issue(args.attachment[0], 'updatedby', updatedby)81 update_issue(args.attachment[0], 'updatedtime', date) 82 filename = ntpath.basename(args.attachment[1]) 83 log_history(args.attachment[0], date, updatedby, 'Attached file: %s/%s/%s.%s' % (attach_location, args.attachment[0], date[0:16], filename))84 comment_issue(args.attachment[0], date, updatedby, 'Attached file: %s/%s/%s.%s' % (attach_location, args.attachment[0], date[0:16], filename), UserGroup)85 view_tracker_update(args.attachment[0],UserGroup) 86 exit(0)87elif '--list' in sys.argv[1]:88 # ./ctt.py --list # Shows all open89 # ./ctt.py --list -s closed # Options: open, closed, deleted90 parser = argparse.ArgumentParser(add_help=False, description="Cluster Ticket Tracker Version 1.0.0")91 parser.add_argument('--list', action='store_true', dest='listvalue', default=True, required=True)92 parser.add_argument('-s', action='store', dest='statusvalue', choices=('open', 'closed', 'deleted', 'all'), required=False)93 parser.add_argument('-v', action='store_true', dest='verbosevalue', default=False, required=False)94 parser.add_argument('-vv', action='store_true', dest='vverbosevalue', default=False, required=False)95 args = parser.parse_args()96 97 if not args.statusvalue:98 args.statusvalue = 'open'99 if args.verbosevalue is True:100 get_issues_v(args.statusvalue)101 exit(0)102 if args.vverbosevalue is True:103 get_issues_vv(args.statusvalue)104 exit(0)105 else:106 get_issues(args.statusvalue)107 exit(0) 108elif '--show' in sys.argv[1]: #We want to see deleted issues as well!!!, FIX109 # ./ctt.py --show 1045110 parser = argparse.ArgumentParser(add_help=False, description="Cluster Ticket Tracker Version 1.0.0")111 parser.add_argument('--show', action='store', dest='issuenumber', nargs=1, required=True) 112 parser.add_argument('-d', action='store_true', default=False)113 args = parser.parse_args()114 if args.issuenumber[0]:115 get_issue_full(args.issuenumber[0])116 if args.d is True:117 get_history(args.issuenumber[0])118 view_tracker_update(args.issuenumber[0],UserGroup)119elif '--update' in sys.argv[1]:120 # ./ctt.py --update 1039 -s 1 -c cheyenne -n r1i1n1 -t 689725 -a casg121 parser = argparse.ArgumentParser(add_help=False, description="Cluster Ticket Tracker Version 1.0.0")122 parser.add_argument('--update', action='store', dest='issuenumber', nargs=1, required=True) 123 parser.add_argument('-s','--severity', action='store', dest='severityvalue', choices=('1','2','3','4'), required=False)124 parser.add_argument('-c','--cluster', action='store', dest='clustervalue', required=False)125 parser.add_argument('-n','--node', action='store', dest='nodevalue', required=False)126 parser.add_argument('-t','--ticket', action='store', dest='ticketvalue', required=False)127 parser.add_argument('-a','--assign', action='store', dest='assignedtovalue', required=False)128 parser.add_argument('-i','--issuetitle', action='store', dest='issuetitlevalue', required=False)129 parser.add_argument('-d','--description', action='store', dest='descvalue', required=False) 130 parser.add_argument('-x', '--type', action='store', dest='typevalue', choices=('h','h!', 's', 't', 'u', 'o'), required=False)131 args = parser.parse_args()132 issue_list = args.issuenumber[0].split(',')133 for cttissue in issue_list:134 if args.typevalue:135 try:136 if cttissue:137 if args.typevalue == 'h!':138 #print(cttissue)139 add_siblings(cttissue,date,updatedby)140 args.typevalue = 'h'141 update_issue(cttissue, 'issuetype', args.typevalue) # 1009 issuetype {hardware,software,test,unknown,other}142 update_issue(cttissue, 'updatedby', updatedby)143 update_issue(cttissue, 'updatedtime', date)144 view_tracker_new(cttissue,UserGroup,viewnotices)145 log_history(cttissue,date,updatedby,'updated issue type to: %s' % (args.typevalue))146 except IndexError:147 parser.print_help()148 if args.issuetitlevalue:149 test_arg_size(args.issuetitlevalue,what='issue title',maxchars=100)150 try:151 if cttissue:152 update_issue(cttissue, 'issuetitle', args.issuetitlevalue)153 update_issue(cttissue, 'updatedby', updatedby)154 update_issue(cttissue, 'updatedtime', date)155 view_tracker_new(cttissue,UserGroup,viewnotices)156 log_history(cttissue,date,updatedby,'updated issue title to: %s' % (args.issuetitlevalue))157 except IndexError:158 parser.print_help()159 if args.descvalue:160 test_arg_size(args.descvalue,what='issue description',maxchars=4000)161 try:162 if cttissue:163 update_issue(cttissue, 'issuedescription', args.descvalue)164 update_issue(cttissue, 'updatedby', updatedby)165 update_issue(cttissue, 'updatedtime', date)166 view_tracker_new(cttissue,UserGroup,viewnotices)167 log_history(cttissue,date,updatedby,'updated issue description to: %s' % (args.descvalue))168 except IndexError:169 parser.print_help()170 if args.severityvalue:171 try:172 if cttissue:173 update_issue(cttissue, 'severity', args.severityvalue)174 update_issue(cttissue, 'updatedby', updatedby)175 update_issue(cttissue, 'updatedtime', date)176 view_tracker_new(cttissue,UserGroup,viewnotices)177 log_history(cttissue,date,updatedby,'updated issue severity to: %s' % (args.severityvalue))178 except IndexError:179 parser.print_help()180 if args.clustervalue:181 try:182 if cttissue:183 update_issue(cttissue, 'cluster', args.clustervalue)184 update_issue(cttissue, 'updatedby', updatedby)185 update_issue(cttissue, 'updatedtime', date)186 view_tracker_new(cttissue,UserGroup,viewnotices)187 log_history(cttissue,date,updatedby,'updated cluster to: %s' % (args.clustervalue))188 except IndexError:189 parser.print_help()190 if args.nodevalue:191 try:192 if cttissue:193 update_issue(cttissue, 'node', args.nodevalue)194 update_issue(cttissue, 'updatedby', updatedby)195 update_issue(cttissue, 'updatedtime', date)196 view_tracker_new(cttissue,UserGroup,viewnotices)197 log_history(cttissue,date,updatedby,'updated node to: %s' % (args.nodevalue))198 except IndexError:199 parser.print_help()200 if args.assignedtovalue in groupsList:201 try:202 if cttissue:203 update_issue(cttissue, 'assignedto', args.assignedtovalue)204 update_issue(cttissue, 'updatedby', updatedby)205 update_issue(cttissue, 'updatedtime', date)206 view_tracker_new(cttissue,UserGroup,viewnotices)207 log_history(cttissue,date,updatedby,'assigned issue to: %s' % (args.assignedtovalue))208 except IndexError:209 parser.print_help()210 elif args.assignedtovalue != None:211 print("Assign to group \"%s\" is not a valid users group, Exiting!" % (args.assignedtovalue))212 if args.ticketvalue:213 try:214 if cttissue:215 update_ticket(cttissue, args.ticketvalue)216 update_issue(cttissue, 'updatedby', updatedby)217 update_issue(cttissue, 'updatedtime', date)218 view_tracker_new(cttissue,UserGroup,viewnotices)219 log_history(cttissue,date,updatedby,'toggled ticket: %s' % (args.ticketvalue))220 except IndexError:221 parser.print_help()222elif '--comment' in sys.argv[1]: 223 # ./ctt.py --comment 12390,12011 "Need an update"224 parser = argparse.ArgumentParser(add_help=False, description="Cluster Ticket Tracker Version 1.0.0")225 parser.add_argument('--comment', action='store', dest='comment', nargs=2, required=True) 226 args = parser.parse_args()227 if args.comment[0] and args.comment[1]:228 issue_list = args.comment[0].split(',')229 for cttissue in issue_list: 230 test_arg_size(args.comment[1],what='comment',maxchars=500)231 comment_issue(cttissue, date, updatedby, args.comment[1],UserGroup)232 update_issue(cttissue, 'updatedby', updatedby)233 update_issue(cttissue, 'updatedtime', date)234 log_history(cttissue, date, updatedby, 'commented issue with: %s' % (args.comment[1]))235elif '--delete' in sys.argv[1]: #make where must be 'admin' to delete 236 # ./ctt.py --delete 10101 "Duplicate issue"237 parser = argparse.ArgumentParser(add_help=False, description="Cluster Ticket Tracker Version 1.0.0")238 parser.add_argument('--delete', action='store', dest='deletevalue', nargs=2, required=True) #NEED TO FIX DELETE 239 args = parser.parse_args()240 if args.deletevalue[0] and args.deletevalue[1]:241 test_arg_size(args.deletevalue[1],what='comment',maxchars=500)242 comment_issue(args.deletevalue[0], date, updatedby, args.deletevalue[1],UserGroup)243 delete_issue(args.deletevalue[0])244 update_issue(args.deletevalue[0], 'updatedby', updatedby)245 update_issue(args.deletevalue[0], 'updatedtime', date)246 log_history(args.deletevalue[0],date,updatedby,'deleted issue: %s' % (args.deletevalue[1]))247elif '--close' in sys.argv[1]:248 # ./ctt.py --close 1028,1044 "Issue resolved"249 parser = argparse.ArgumentParser(add_help=False, description="Cluster Ticket Tracker Version 1.0.0")250 parser.add_argument('--close', action='store', dest='closevalue', nargs=2, required=True) 251 args = parser.parse_args()252 if args.closevalue[0] and args.closevalue[1]:253 issue_list = args.closevalue[0].split(',')254 for cttissue in issue_list:255 test_arg_size(args.closevalue[1],what='comment',maxchars=500)256 comment_issue(cttissue, date, updatedby, args.closevalue[1],UserGroup)257 update_issue(cttissue, 'updatedby', updatedby)258 update_issue(cttissue, 'updatedtime', date)259 close_issue(cttissue, date, updatedby)260 log_history(cttissue, date, updatedby, 'closed issue: %s' % (args.closevalue[1]))261elif '--reopen' in sys.argv[1]:262 # ./ctt.py --reopen 10282,10122 "Accidental close"263 parser = argparse.ArgumentParser(add_help=False, description="Cluster Ticket Tracker Version 1.0.0")264 parser.add_argument('--reopen', action='store', dest='reopenvalue', nargs=2, required=True) 265 args = parser.parse_args()266 if args.reopenvalue[0] and args.reopenvalue[1]:267 issue_list = args.reopenvalue[0].split(',')268 for cttissue in issue_list:269 test_arg_size(args.reopenvalue[1],what='comment',maxchars=500)270 comment_issue(cttissue, date, updatedby, args.reopenvalue[1],UserGroup)271 update_issue(cttissue, 'updatedby', updatedby)272 update_issue(cttissue, 'updatedtime', date)273 update_issue(cttissue, 'status', 'open')274 log_history(cttissue, date, updatedby, 'reopened issue: %s' % (args.reopenvalue[1]))275 if pbs_enforcement == 'False':276 print("pbs_enforcement is False. Not draining nodes")277 else:278 pbs_drain(cttissue,date,updatedby,get_hostname(cttissue))279elif '--open' in sys.argv[1]:280 # ./ctt.py --open "Failed dimm on r1i1n1" "Description here" -c cheyenne -s 1 -n r1i1n1,r1i1n10 -a casg281 parser = argparse.ArgumentParser(add_help=False, description="Cluster Ticket Tracker Version 1.0.0")282 parser.add_argument('--open', action='store', dest='openvalue', nargs='+', required=True)283 parser.add_argument('-s','--severity', action='store', dest='severityvalue', required=False)284 parser.add_argument('-c','--cluster', action='store', dest='clustervalue', required=False)285 parser.add_argument('-n','--node', action='store', dest='nodevalue', required=True)286 parser.add_argument('-a','--assign', action='store', dest='assignedtovalue', required=False)287 parser.add_argument('-t', '--ticket', action='store', dest='ticketvalue', required=False)...
test_celery_tasks.py
Source:test_celery_tasks.py
1from collections import defaultdict2from django.test import TestCase3from unittest import mock4from device_registry.models import GithubIssue5from device_registry.recommended_actions import ActionMeta, SimpleAction, Severity6from profile_page.models import *7TEST_REPO_ID = 12348TEST_ISSUE_ID = 23459def repo_mock():10 class GithubRepoMock:11 list_issues = mock.Mock()12 close_issue = mock.Mock()13 update_issue = mock.Mock()14 create_issue = mock.Mock()15 def __init__(self, repo_id):16 self.repo_id = repo_id17 return GithubRepoMock18@mock.patch('device_registry.celery_tasks.github.list_repos', lambda t: {TEST_REPO_ID: {}})19@mock.patch('device_registry.celery_tasks.github.GithubRepo', new_callable=repo_mock)20class GithubTest(TestCase):21 def setUp(self):22 hour_ago = timezone.now() - timezone.timedelta(hours=1)23 self.user = User.objects.create_user('test')24 self.profile = Profile.objects.create(user=self.user, github_oauth_token='abcd', github_repo_id=1234)25 self.device = Device.objects.create(26 device_id='device.d.wott-dev.local',27 last_ping=hour_ago,28 owner=self.user,29 name='testdevice'30 )31 self.device1 = Device.objects.create(32 device_id='device1.d.wott-dev.local',33 last_ping=hour_ago,34 owner=self.user,35 name='testdevice1'36 )37 @classmethod38 def setUpClass(cls):39 super().setUpClass()40 class TestAction(SimpleAction, metaclass=ActionMeta):41 action_config = defaultdict(str)42 _severity = Severity.LO43 cls.TestAction = TestAction44 @classmethod45 def tearDownClass(cls):46 super().tearDownClass()47 ActionMeta.unregister(cls.TestAction)48 def test_empty(self, mockGithubRepo):49 self.device.owner = None50 self.device.save()51 # No devices - do nothing52 self.assertEqual(github.file_issues(), 0)53 def test_empty_ra(self, mockGithubRepo):54 mockGithubRepo.list_issues.return_value = {}55 self.assertEqual(github.file_issues(), 0)56 def test_create_ra(self, mockGithubRepo):57 ra = RecommendedAction.objects.create(action_class=self.TestAction.__name__, action_param='param')58 RecommendedActionStatus.objects.create(device=self.device, status=RecommendedAction.Status.AFFECTED, ra=ra)59 mockGithubRepo.list_issues.return_value = {}60 mockGithubRepo.create_issue.return_value = TEST_ISSUE_ID61 self.assertEqual(github.file_issues(), 1)62 title, text = mockGithubRepo.create_issue.call_args[0]63 self.assertIn('#### Resolved on: ####\n- [ ] [testdevice]', text)64 issue = GithubIssue.objects.get(number=TEST_ISSUE_ID, owner=self.user)65 self.assertFalse(issue.closed)66 def test_update_ra_open(self, mockGithubRepo):67 ra = RecommendedAction.objects.create(action_class=self.TestAction.__name__, action_param='param')68 RecommendedActionStatus.objects.create(device=self.device, status=RecommendedAction.Status.AFFECTED, ra=ra)69 issue = GithubIssue.objects.create(number=TEST_ISSUE_ID, owner=self.user, ra=ra)70 mockGithubRepo.list_issues.return_value = {71 'open': [TEST_ISSUE_ID],72 'closed': []73 }74 mockGithubRepo.update_issue.assert_not_called()75 self.assertEqual(github.file_issues(), 1)76 mockGithubRepo.update_issue.assert_called_once()77 number, text = mockGithubRepo.update_issue.call_args[0]78 self.assertEqual(number, TEST_ISSUE_ID)79 self.assertIn('#### Resolved on: ####\n- [ ] [testdevice]', text)80 issue.refresh_from_db()81 self.assertFalse(issue.closed)82 # Now make sure it doesn't happen twice83 self.assertEqual(github.file_issues(), 1)84 mockGithubRepo.update_issue.assert_called_once()85 def test_update_ra_closed(self, mockGithubRepo):86 ra = RecommendedAction.objects.create(action_class=self.TestAction.__name__, action_param='param')87 RecommendedActionStatus.objects.create(device=self.device, status=RecommendedAction.Status.AFFECTED, ra=ra)88 issue = GithubIssue.objects.create(number=TEST_ISSUE_ID, owner=self.user, closed=True, ra=ra)89 mockGithubRepo.list_issues.return_value = {90 'open': [],91 'closed': [TEST_ISSUE_ID]92 }93 mockGithubRepo.update_issue.assert_not_called()94 self.assertEqual(github.file_issues(), 1)95 mockGithubRepo.update_issue.assert_called_once()96 number, text = mockGithubRepo.update_issue.call_args[0]97 self.assertEqual(number, TEST_ISSUE_ID)98 self.assertIn('#### Resolved on: ####\n- [ ] [testdevice]', text)99 issue.refresh_from_db()100 self.assertFalse(issue.closed)101 # Now make sure it doesn't happen twice102 mockGithubRepo.list_issues.return_value = {103 'open': [TEST_ISSUE_ID],104 'closed': []105 }106 self.assertEqual(github.file_issues(), 1)107 mockGithubRepo.update_issue.assert_called_once()108 def test_resolve_ra(self, mockGithubRepo):109 ra = RecommendedAction.objects.create(action_class=self.TestAction.__name__, action_param='param')110 RecommendedActionStatus.objects.create(device=self.device, status=RecommendedAction.Status.NOT_AFFECTED,111 resolved_at=timezone.now(), ra=ra)112 issue = GithubIssue.objects.create(number=TEST_ISSUE_ID, owner=self.user, ra=ra)113 mockGithubRepo.list_issues.return_value = {114 'open': [TEST_ISSUE_ID],115 'closed': []116 }117 self.assertEqual(github.file_issues(), 1)118 number = mockGithubRepo.close_issue.call_args[0][0]119 self.assertEqual(number, TEST_ISSUE_ID)120 issue.refresh_from_db()121 self.assertTrue(issue.closed)122 # Now make sure it doesn't happen twice123 mockGithubRepo.list_issues.return_value = {124 'open': [],125 'closed': [TEST_ISSUE_ID]126 }127 self.assertEqual(github.file_issues(), 0)128 mockGithubRepo.close_issue.assert_called_once()129 def test_partly_resolve_ra(self, mockGithubRepo):130 ra = RecommendedAction.objects.create(action_class=self.TestAction.__name__, action_param='param')131 RecommendedActionStatus.objects.create(device=self.device, status=RecommendedAction.Status.AFFECTED,132 ra=ra)133 RecommendedActionStatus.objects.create(device=self.device1, status=RecommendedAction.Status.NOT_AFFECTED,134 resolved_at=timezone.now(), ra=ra)135 issue = GithubIssue.objects.create(number=TEST_ISSUE_ID, owner=self.user, ra=ra)136 mockGithubRepo.list_issues.return_value = {137 'open': [TEST_ISSUE_ID],138 'closed': []139 }140 self.assertEqual(github.file_issues(), 1)141 number, text = mockGithubRepo.update_issue.call_args[0]142 self.assertEqual(number, TEST_ISSUE_ID)143 self.assertIn('\n- [ ] [testdevice]', text)144 self.assertIn('\n- [x] [testdevice1]', text)145 issue.refresh_from_db()...
test_sentry_apps.py
Source:test_sentry_apps.py
...16 self.url = u"/api/0/projects/{}/{}/issues/?id={}".format(17 self.organization.slug, self.issue.project.slug, self.issue.id18 )19 self.login_as(self.user)20 def update_issue(self, _data=None):21 data = {"status": "resolved"}22 data.update(_data or {})23 self.client.put(self.url, data=data, format="json")24 def test_notify_after_basic_resolved(self, delay):25 self.update_issue()26 assert faux(delay).called_with(27 installation_id=self.install.id,28 issue_id=self.issue.id,29 type="resolved",30 user_id=self.user.id,31 data={"resolution_type": "now"},32 )33 def test_notify_after_resolve_in_commit(self, delay):34 repo = self.create_repo(project=self.project)35 commit = self.create_commit(repo=repo)36 self.update_issue(37 {"statusDetails": {"inCommit": {"repository": repo.name, "commit": commit.key}}}38 )39 assert faux(delay).called_with(40 installation_id=self.install.id,41 issue_id=self.issue.id,42 type="resolved",43 user_id=self.user.id,44 data={"resolution_type": "in_commit"},45 )46 def test_notify_after_resolve_in_specific_release(self, delay):47 release = self.create_release(project=self.project)48 self.update_issue({"statusDetails": {"inRelease": release.version}})49 assert faux(delay).called_with(50 installation_id=self.install.id,51 issue_id=self.issue.id,52 type="resolved",53 user_id=self.user.id,54 data={"resolution_type": "in_release"},55 )56 def test_notify_after_resolve_in_latest_release(self, delay):57 self.create_release(project=self.project)58 self.update_issue({"statusDetails": {"inRelease": "latest"}})59 assert faux(delay).called_with(60 installation_id=self.install.id,61 issue_id=self.issue.id,62 type="resolved",63 user_id=self.user.id,64 data={"resolution_type": "in_release"},65 )66 def test_notify_after_resolve_in_next_release(self, delay):67 self.create_release(project=self.project)68 self.update_issue({"statusDetails": {"inNextRelease": True}})69 assert faux(delay).called_with(70 installation_id=self.install.id,71 issue_id=self.issue.id,72 type="resolved",73 user_id=self.user.id,74 data={"resolution_type": "in_next_release"},75 )76 def test_notify_after_resolve_from_set_commits(self, delay):77 repo = Repository.objects.create(organization_id=self.organization.id, name="test/repo")78 release = Release.objects.create(version="abcabc", organization=self.organization)79 commit = Commit.objects.create(80 repository_id=repo.id, organization_id=self.organization.id, key="b" * 4081 )82 GroupLink.objects.create(83 group_id=self.issue.id,84 project_id=self.project.id,85 linked_type=GroupLink.LinkedType.commit,86 linked_id=commit.id,87 )88 release.add_project(self.project)89 release.set_commits(90 [91 {92 "id": "b" * 40,93 "repository": repo.name,94 "author_email": "foo@example.com",95 "author_name": "Foo Bar",96 "message": u"FIXES {}".format(self.issue.qualified_short_id),97 }98 ]99 )100 assert faux(delay).called_with(101 installation_id=self.install.id,102 issue_id=self.issue.id,103 type="resolved",104 user_id=None,105 data={"resolution_type": "with_commit"},106 )107 def test_notify_after_issue_ignored(self, delay):108 self.update_issue({"status": "ignored"})109 assert faux(delay).called_with(110 installation_id=self.install.id,111 issue_id=self.issue.id,112 type="ignored",113 user_id=self.user.id,114 data={},115 )116@patch("sentry.tasks.sentry_apps.workflow_notification.delay")117class TestIssueAssigned(APITestCase):118 def setUp(self):119 self.issue = self.create_group(project=self.project)120 self.sentry_app = self.create_sentry_app(events=["issue.assigned"])121 self.install = self.create_sentry_app_installation(122 organization=self.organization, slug=self.sentry_app.slug...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!