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

How to use transcribe_create_job method in localstack

Best Python code snippet using localstack_python

fixtures.py

Source:fixtures.py

...539            route53_client.delete_hosted_zone(Id=zone)540        except Exception as e:541            LOG.debug(f"error cleaning up route53 HostedZone {zone}: {e}")542@pytest.fixture543def transcribe_create_job(transcribe_client, s3_client, s3_bucket):544    def _create_job(audio_file: str, params: Optional[dict[str, Any]] = None) -> str:545        s3_key = "test-clip.wav"546        if not params:547            params = {}548        if "TranscriptionJobName" not in params:549            params["TranscriptionJobName"] = f"test-transcribe-{short_uid()}"550        if "LanguageCode" not in params:551            params["LanguageCode"] = "en-GB"552        if "Media" not in params:553            params["Media"] = {"MediaFileUri": f"s3://{s3_bucket}/{s3_key}"}554        # upload test wav to a s3 bucket555        with open(audio_file, "rb") as f:556            s3_client.upload_fileobj(f, s3_bucket, s3_key)557        transcribe_client.start_transcription_job(**params)...

test_transcribe.py

Source:test_transcribe.py

...24        ]25    )26    def test_transcribe_happy_path(self, transcribe_client, transcribe_create_job, snapshot):27        file_path = os.path.join(BASEDIR, "files/en-gb.wav")28        job_name = transcribe_create_job(audio_file=file_path)29        transcribe_client.get_transcription_job(TranscriptionJobName=job_name)30        def is_transcription_done():31            transcription_status = transcribe_client.get_transcription_job(32                TranscriptionJobName=job_name33            )34            return transcription_status["TranscriptionJob"]["TranscriptionJobStatus"] == "COMPLETED"35        # empirically it takes around36        # <5sec for a vosk transcription37        # ~100sec for an AWS transcription -> adjust timeout accordingly38        assert poll_condition(39            is_transcription_done, timeout=10040        ), f"could not finish transcription job: {job_name} in time"41        job = transcribe_client.get_transcription_job(TranscriptionJobName=job_name)42        snapshot.match("TranscriptionJob", job)43        # delete the job again44        transcribe_client.delete_transcription_job(TranscriptionJobName=job_name)45        # check if job is gone46        with pytest.raises((ClientError, NotFoundException)) as e_info:47            transcribe_client.get_transcription_job(TranscriptionJobName=job_name)48        snapshot.match("GetError", e_info.value.response)49    @pytest.mark.aws_validated50    @pytest.mark.skip_snapshot_verify(51        paths=["$..TranscriptionJob..Settings", "$..TranscriptionJob..Transcript", "$..Error..Code"]52    )53    def test_get_transcription_job(self, transcribe_client, transcribe_create_job, snapshot):54        file_path = os.path.join(BASEDIR, "files/en-gb.wav")55        job_name = transcribe_create_job(audio_file=file_path)56        job = transcribe_client.get_transcription_job(TranscriptionJobName=job_name)57        snapshot.match("GetJob", job)58        with pytest.raises((ClientError, NotFoundException)) as e_info:59            transcribe_client.get_transcription_job(TranscriptionJobName="non-existent")60        snapshot.match("GetError", e_info.value.response)61    @pytest.mark.aws_validated62    @pytest.mark.skip_snapshot_verify(63        paths=["$..NextToken", "$..TranscriptionJobSummaries..OutputLocationType"]64    )65    def test_list_transcription_jobs(self, transcribe_client, transcribe_create_job, snapshot):66        file_path = os.path.join(BASEDIR, "files/en-gb.wav")67        transcribe_create_job(audio_file=file_path)68        jobs = transcribe_client.list_transcription_jobs()69        # there are potentially multiple transcription jobs on AWS - ordered by creation date70        # we only care about the newest one that we just created71        jobs["TranscriptionJobSummaries"] = jobs["TranscriptionJobSummaries"][0]72        snapshot.match("ListJobs", jobs)73    @pytest.mark.aws_validated74    @pytest.mark.skip_snapshot_verify(paths=["$..Error..Code"])75    def test_failing_deletion(self, transcribe_client, snapshot):76        # successful deletion is tested in the happy path test77        # this tests a failed deletion78        with pytest.raises((ClientError, NotFoundException)) as e_info:79            transcribe_client.delete_transcription_job(TranscriptionJobName="non-existent")80        snapshot.match("MissingLanguageCode", e_info.value.response)81    @pytest.mark.aws_validated...

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.