KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use validate_attachments method in mailosaur-python

Best Python code snippet using mailosaur-python_python

test_integration.py

Source:test_integration.py

...91    if validator.same_columns_in_ref_and_test() and validator.all_test_columns_in_ref_and_test() and validator.check_samples_present() and validator.check_columns_match():92        return True93    else:94        return False95def validate_attachments(issue, file_to_find, validate_content=False):96    validation_status = 'Unknown'97    if len(issue.attachments) > 0:98        attachment_found = False99        for attachment in issue.attachments:100            if attachment.filename == file_to_find:101                attachment_found = True102                with tempfile.TemporaryDirectory() as tmpdir:103                    attachment.download(savepath=tmpdir, filename=file_to_find)104                    if os.path.getsize(os.path.join(tmpdir, file_to_find)) > 0:105                        if validate_content is True:106                            ref_csv = 'tests/ref_csvs/{}'.format(file_to_find)107                            content_ok = validate_csv_content(query_csv=os.path.join(tmpdir, file_to_find),108                                                              ref_csv='tests/ref_csvs/{}'.format(file_to_find))109                            if content_ok is True:110                                validation_status = 'Validated'111                            else:112                                validation_status = 'Reference content does not match query.'113                        else:114                                validation_status = 'Validated'115                    else:116                        validation_status = 'Uploaded file has zero size.'117        if attachment_found is False:118            validation_status = 'Could not find {}'.format(file_to_find)119    else:120        validation_status = 'No attachments, validation failed.'121    return validation_status122def md5(fname):123    """124    Gets MD5sum for a file - use for quick and dirty comparisons of files.125    Hooray for stackoverflow: https://stackoverflow.com/questions/3431825/generating-an-md5-checksum-of-a-file126    """127    hash_md5 = hashlib.md5()128    with open(fname, "rb") as f:129        for chunk in iter(lambda: f.read(4096), b""):130            hash_md5.update(chunk)131    return hash_md5.hexdigest()132def validate_csv_in_zip(issue, zip_file, report_file, ref_csv):133    """134    Crude validation that checks that CSV contents are the exact same between some folder in a zip135    file and a reference that we keep on hand.136    :param issue:137    :param zip_file:138    :param report_file:139    :param ref_csv:140    :return:141    """142    validation_status = 'Unknown'143    if len(issue.attachments) > 0:144        attachment_found = False145        for attachment in issue.attachments:146            if attachment.filename == zip_file:147                attachment_found = True148                with tempfile.TemporaryDirectory() as tmpdir:149                    downloaded_zip = os.path.join(tmpdir, zip_file)150                    attachment.download(savepath=tmpdir, filename=zip_file)151                    zipped_archive = zipfile.ZipFile(downloaded_zip)152                    zipped_archive.extractall(path=tmpdir)153                    if md5(ref_csv) == md5(os.path.join(tmpdir, report_file)):154                        validation_status = 'Validated'155                    else:156                        validation_status = 'Contents not identical. Check on what changed.'157        if attachment_found is False:158            validation_status = 'Could not find {}'.format(zip_file)159    else:160        validation_status = 'No attachments, validation failed.'161    return validation_status162def validate_ftp_upload(ftp_file, min_file_size=0):163    try:164        with tempfile.TemporaryDirectory() as tmpdir:165            urllib.request.urlretrieve(os.path.join(OUTGOING_FTP, ftp_file),166                                       os.path.join(tmpdir, ftp_file))167            if os.path.getsize(os.path.join(tmpdir, ftp_file)) > min_file_size:168                validation_status = 'Validated'169            else:170                validation_status = 'Uploaded file has zero size.'171    except:  # If file doesn't exist, something went wrong. Set validation status accordingly.172        validation_status = 'No uploaded file found on FTP'173    return validation_status174def monitor_issues(redmine, issue_dict, timeout):175    """176    Monitors issues for completion.177    :param redmine: Instantiated redmine instance.178    :param issue_dict: Dictionary created by create_test_issues (issue subjects as keys, issue IDs as values)179    :param timeout: Number of seconds to wait for all issues to finish running180    :return: Dictionary with issue subjects as keys, and some sort of status message for each as values181    """182    all_complete = False183    total_time_taken = 0184    increment = 30  # Check on issues every increment seconds185    # Create dictionary to track progress of each issue created.186    issues_validated = dict()187    for issue_subject in issue_dict:188        issues_validated[issue_subject] = 'Unknown'189    # Now loop through all the things!190    while all_complete is False and total_time_taken < timeout:191        logging.info('Checking tasks for completion')192        all_complete = True  # Set flag to true - if any of our issues are not finished, this gets set back to False193        # Check all issues for completion.194        for issue_subject in issue_dict:195            issue_id = issue_dict[issue_subject]196            issue = redmine.issue.get(issue_id, include=['attachments'])197            # AMR Summary #####198            if issue_subject == 'amrsummary' and issues_validated[issue_subject] == 'Unknown':199                if issue.status.id == 4:  # 4 corresponds to issue being complete.200                    issues_validated[issue_subject] = validate_attachments(issue, 'amr_summary.csv',201                                                                           validate_content=True)202                    logging.info('{} is complete, status is {}'.format(issue_subject, issues_validated[issue_subject]))203                else:  # If hasn't finished yet, just wait!204                    all_complete = False205            # CLARK #####206            elif issue_subject == 'autoclark' and issues_validated[issue_subject] == 'Unknown':207                if issue.status.id == 4:208                    # TODO: Validate xlsx files?209                    issues_validated[issue_subject] = validate_attachments(issue, 'abundance.xlsx')210                    logging.info('{} is complete, status is {}'.format(issue_subject, issues_validated[issue_subject]))211                else:212                    all_complete = False213            # ECGF #####214            elif issue_subject == 'ecgf' and issues_validated[issue_subject] == 'Unknown':215                if issue.status.id == 4:216                    issues_validated[issue_subject] = validate_csv_in_zip(issue=issue,217                                                                          zip_file='eCGF_output_{}.zip'.format(issue.id),218                                                                          report_file='{}_summary_report.csv'.format(issue.id),219                                                                          ref_csv='/mnt/nas2/redmine/applications/OLCRedmineAutomator/tests/ref_csvs/124_summary_report.csv')  # Hardcoded paths stink, fix me!220                    logging.info('{} is complete, status is {}'.format(issue_subject, issues_validated[issue_subject]))221                else:222                    all_complete = False223            # EC TYPER #####224            elif issue_subject == 'ec_typer' and issues_validated[issue_subject] == 'Unknown':225                if issue.status.id == 4:226                    issues_validated[issue_subject] = validate_attachments(issue, 'ec_typer_report.tsv',227                                                                           validate_content=True)228                    logging.info('{} is complete, status is {}'.format(issue_subject, issues_validated[issue_subject]))229                else:230                    all_complete = False231            # EXTERNAL RETRIEVE #####232            elif issue_subject == 'externalretrieve' and issues_validated[issue_subject] == 'Unknown':233                if issue.status.id == 4:234                    issues_validated[issue_subject] = validate_ftp_upload('{}.zip'.format(issue.id),235                                                                          min_file_size=1000)236                else:237                    all_complete = False238            # DIVERSITREE #####239            elif issue_subject == 'diversitree' and issues_validated[issue_subject] == 'Unknown':240                if issue.status.id == 4:241                    # TODO: HTML validation - should just be able to md5sum or something on file contents.242                    issues_validated[issue_subject] = validate_attachments(issue, 'diversitree_report.html')243                    logging.info('{} is complete, status is {}'.format(issue_subject, issues_validated[issue_subject]))244                else:245                    all_complete = False246            # GENESEEKR #####247            elif issue_subject == 'geneseekr' and issues_validated[issue_subject] == 'Unknown':248                if issue.status.id == 4:249                    issues_validated[issue_subject] = validate_csv_in_zip(issue=issue,250                                                                          zip_file='geneseekr_output.zip',251                                                                          report_file='2014-SEQ-0276_blastn_sixteens_full.tsv',252                                                                          ref_csv='/mnt/nas2/redmine/applications/OLCRedmineAutomator/tests/ref_csvs/2014-SEQ-0276_blastn_sixteens_full.tsv')  # Hardcoded paths stink, fix me!253                    logging.info('{} is complete, status is {}'.format(issue_subject, issues_validated[issue_subject]))254                else:255                    all_complete = False256            # POINTFINDER #####257            elif issue_subject == 'pointfinder' and issues_validated[issue_subject] == 'Unknown':258                if issue.status.id == 4:259                    issues_validated[issue_subject] = validate_csv_in_zip(issue=issue,260                                                                          zip_file='pointfinder_output.zip',261                                                                          report_file='PointFinder_results_summary.csv',262                                                                          ref_csv='/mnt/nas2/redmine/applications/OLCRedmineAutomator/tests/ref_csvs/PointFinder_results_summary.csv')  # Hardcoded paths stink, fix me!263                    logging.info('{} is complete, status is {}'.format(issue_subject, issues_validated[issue_subject]))264                else:265                    all_complete = False266            # PROKKA #####267            elif issue_subject == 'prokka' and issues_validated[issue_subject] == 'Unknown':268                if issue.status.id == 4:269                    issues_validated[issue_subject] = validate_ftp_upload('prokka_output_{}.zip'.format(issue.id),270                                                                          min_file_size=1000)271                else:272                    all_complete = False273            # RESFINDER #####274            elif issue_subject == 'resfinder' and issues_validated[issue_subject] == 'Unknown':275                if issue.status.id == 4:276                    issues_validated[issue_subject] = validate_attachments(issue, 'resfinder_blastn.xlsx')277                    logging.info('{} is complete, status is {}'.format(issue_subject, issues_validated[issue_subject]))278                else:279                    all_complete = False280            # SIPPRVERSE #####281            elif issue_subject == 'sipprverse' and issues_validated[issue_subject] == 'Unknown':282                if issue.status.id == 4:283                    issues_validated[issue_subject] = validate_csv_in_zip(issue=issue,284                                                                          zip_file='sipprverse_output.zip',285                                                                          report_file='mlst.csv',286                                                                          ref_csv='/mnt/nas2/redmine/applications/OLCRedmineAutomator/tests/ref_csvs/mlst.csv')  # Hardcoded paths stink, fix me!287                    logging.info('{} is complete, status is {}'.format(issue_subject, issues_validated[issue_subject]))288                else:289                    all_complete = False290            # STARAMR #####...

middleware.py

Source:middleware.py

...14from .tasks import run_judge15class OJSubmissionSerializer(serializers.Serializer):16    post = serializers.CharField()17    def validate_post(self, content):18        self.validate_attachments()19        starter = self.context['starter']20        thread = self.context['thread']21        uid = self.context['uid']22        ps = Problem.objects.filter(uid=UUID(uid))23        if ps.count() == 0:24            raise serializers.ValidationError(_('No such problem'))25        problem = ps[0]26        self.problem = problem27        sol = Solution.objects.filter(problem=problem, starter=starter)28        self.solution = None29        if sol.exists():30            solution = sol[0]31            if solution.thread != thread:32                raise serializers.ValidationError(_(33                    'Only one solution should be '34                    'submitted to the same problem'))35            self.solution = solution36    def validate_attachments(self):37        ids = self.context['attachments']38        attachments = Attachment.objects.filter(id__in=ids)39        if len(attachments) == 0:40            raise serializers.ValidationError(_('Must provide a attachment'))41        elif len(attachments) != 1:42            raise serializers.ValidationError(43                _('Only one attachment required'))44        upfile = attachments[0]45        if str(upfile.filetype) != 'ZIP':46            raise serializers.ValidationError(_('Only support ZIP file'))47        self.subject_path = osp.join(settings.MEDIA_ROOT, str(upfile.file))48    def post_save(self):49        problem = self.problem50        solution = self.solution51        starter = self.context['starter']52        thread = self.context['thread']53        if solution is None:54            solution = Solution.objects.create(problem=problem,55                                               starter=starter,56                                               thread=thread)57        solution.submit_counter = F('submit_counter') + 158        solution.save()59        problem_attachments = problem.thread.first_post.attachments60        target_id = None61        for a in problem_attachments:62            if a['filetype'] == 'SIEZIP':63                target_id = a['id']64                break65        target_path = Attachment.objects.get(id=target_id).file66        target_path = osp.join(settings.MEDIA_ROOT, str(target_path))67        run_judge.delay(thread.id, solution.id, self.subject_path,68                        problem.module_name, target_path, problem.judge_name,69                        problem.order_type)70class OJSubmissionMiddleware(PostingMiddleware):71    def get_serializer(self):72        return OJSubmissionSerializer(73            data=self.request.data,74            context={75                'mode': self.mode,76                'post': self.post,77                'thread': self.thread,78                'starter': self.user,79                'attachments': self.request.data['attachments'],80                'uid': get_problem_uid(self.request.data['post']),81            }82        )83    def use_this_middleware(self):84        if self.mode == PostingEndpoint.EDIT:85            is_sub_cate = str(self.post.category) == settings.OJ_SUB_CATE_NAME86        elif self.mode == PostingEndpoint.START:87            cate_id = self.request.data['category']88            is_sub_cate = Category.objects.filter(89                id=cate_id, name=settings.OJ_SUB_CATE_NAME).count() == 190        else:91            is_sub_cate = False92        uid = get_problem_uid(self.request.data['post'])93        if is_sub_cate and self.post.is_first_post and uid is not None:94            return True95        return False96    def post_save(self, serializer):97        serializer.post_save()98class OJReleaseSerializer(serializers.Serializer):99    post = serializers.CharField()100    def validate_attachments(self):101        ids = self.context['attachments']102        attachments = Attachment.objects.filter(id__in=ids)103        siezip_cnt = 0104        for upfile in attachments:105            if str(upfile.filetype) == 'SIEZIP':106                siezip_cnt += 1107        if siezip_cnt != 1:108            raise serializers.ValidationError(109                _('Requires one SIEZIP file'))110    def validate_post(self, content):111        self.validate_attachments()112        try:113            order_type, judge_name, module_name = extract_problem_kwargs(114                content)115        except Exception as e:116            raise serializers.ValidationError(_('Judge argument error'))117        self.order_type = order_type118        self.judge_name = judge_name119        self.module_name = module_name120    def save(self):121        thread = self.context['thread']122        starter = self.context['user']123        post = self.context['post']124        problem, created = Problem.objects.get_or_create(125            thread=thread, starter=starter)...

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.