ffx/bin/ffx/test/scenario_1.py

import os, sys, click, glob

from .scenario import Scenario

from ffx.test.helper import createMediaTestFile
from ffx.process import executeProcess

from ffx.file_properties import FileProperties

from ffx.media_descriptor import MediaDescriptor
from ffx.track_descriptor import TrackDescriptor

from ffx.track_type import TrackType
from ffx.track_disposition import TrackDisposition

from ffx.test.media_combinator_0 import MediaCombinator0

from ffx.test.basename_combinator import BasenameCombinator


class Scenario1(Scenario):
    """Creating file VAa, h264/aac/aac
    Converting to VaA, vp9/opus/opus
    No tmdb, default parameters"""

    TEST_FILE_EXTENSION = 'mkv'
    EXPECTED_FILE_EXTENSION = 'webm'


    def __init__(self, context):

        context['use_jellyfin'] = True
        context['use_tmdb'] = False
        context['use_pattern'] = False

        super().__init__(context)

    def getScenario(self):
        return self.__class__.__name__[8:]


    def job(self, yieldObj: dict):

        testContext = self._context.copy()

        identifier = yieldObj['identifier']
        variantList = yieldObj['variants']

        variantIdentifier = '-'.join(variantList)
        variantLabel = f"{self.__class__.__name__} Variant {variantIdentifier}"

        mc0 = MediaCombinator0(context = testContext)
        sourceMediaDescriptor: MediaDescriptor = mc0.getPayload()

        assertSelectorList: list = yieldObj['assertSelectors']
        assertFuncList = yieldObj['assertFuncs']
        shouldFail = yieldObj['shouldFail']

        variantPayload = yieldObj['payload']
        variantBasename = variantPayload['basename']
        variantFilenameLabel = variantPayload['label']
        expectedBasename = variantPayload['expectedBasename']

        variantFilename = f"{variantBasename}.{Scenario1.TEST_FILE_EXTENSION}"
        expectedFilename = f"{expectedBasename}.{Scenario1.EXPECTED_FILE_EXTENSION}"


        if self._context['test_variant'] and not variantIdentifier.startswith(self._context['test_variant']):
            return

        if ((self._context['test_passed_counter'] + self._context['test_failed_counter'])
                    >= self._context['test_limit']):
            return

        self._logger.debug(f"Running Job: {variantLabel}")


        # Phase 1: Setup source files

        if not variantBasename:
            raise ValueError(f"{variantLabel}: Testfile basename is falsy")


        self.clearTestDirectory()

        self._logger.debug(f"Creating test file: {variantFilename}")
        mediaFilePath = createMediaTestFile(mediaDescriptor=sourceMediaDescriptor,
                                            baseName=variantBasename,
                                            directory=self._testDirectory,
                                            logger=self._logger,
                                            length = 2)


        # Phase 2: Run ffx

        commandSequence = [sys.executable,
                            self._ffxExecutablePath]
        
        if self._context['verbosity']:
            commandSequence += ['--verbose',
                                str(self._context['verbosity'])]

        commandSequence += ['convert',
                            mediaFilePath,
                            '--no-prompt',
                            '--no-signature']
        
        if variantFilenameLabel:
            commandSequence += ['--label', variantFilenameLabel]


        if not testContext['use_jellyfin']:
            commandSequence += ['--no-jellyfin']

        commandSequence += ['--no-pattern']
        commandSequence += ['--no-tmdb']


        self._logger.debug(f"{variantLabel}: Test sequence: {commandSequence}")

        out, err, rc = executeProcess(commandSequence, directory = self._testDirectory)

        if out and self._context['verbosity'] >= 9:
            self._logger.debug(f"{variantLabel}: Process output: {out}")
        if rc:
            self._logger.debug(f"{variantLabel}: Process returned ERROR {rc} ({err})")


        # Phase 3: Evaluate results
            
        resultFilenames = [rf for rf in self.getFilenamesInTestDirectory() if rf != 'ffmpeg2pass-0.log' and rf != variantFilename]
        self._logger.debug(f"{variantLabel}: Result filenames: {resultFilenames}")


        try:

            jobFailed = bool(rc)

            self._logger.debug(f"{variantLabel}: Should fail: {shouldFail} / actually failed: {jobFailed}")

            assert (jobFailed == shouldFail
            ), f"Process {'failed' if jobFailed else 'did not fail'}"

            if not jobFailed:

                expectedResultFilePath = os.path.join(self._testDirectory, f"{expectedFilename}")

                assert (os.path.isfile(expectedResultFilePath)
                ), f"Result file {expectedFilename} in path '{self._testDirectory}' wasn't created"

                for assertIndex in range(len(assertSelectorList)):

                    assertSelector = assertSelectorList[assertIndex]
                    assertFunc = assertFuncList[assertIndex]
                    assertVariant = variantList[assertIndex]

                    if assertSelector == 'B':
                        #TODO: per file find
                        testObj = {'filenames': resultFilenames}
                        assertFunc(testObj=testObj)
                    if assertSelector == 'L':
                        assertFunc()
                    if assertSelector == 'I':
                        assertFunc()

            self._context['test_passed_counter'] += 1
            self._reportLogger.info(f"{variantLabel}: Test passed")

        except AssertionError as ae:

            self._context['test_failed_counter'] += 1
            self._reportLogger.error(f"{variantLabel}: Test FAILED ({ae})")


    def run(self):
        for BC in BasenameCombinator.getAllClassReferences():
            self._logger.debug(f"BC={BC.__name__}")
            bc = BC(context = self._context)
            for y in bc.getYield():
                self.job(y)