Merge pull request #365 from keboola/erik-SOX-461

ErikZigo · web-flow · commit ccf58752ceca · 2023-12-12T10:54:41.000+01:00
re-run orchestration fixup
diff --git a/src/JobFactory/Job.php b/src/JobFactory/Job.php
@@ -381,19 +381,10 @@ public function getComponentConfiguration(): array
         if ($this->componentConfiguration !== null) {
             return $this->componentConfiguration;
         }
-
-        if (!$this->getConfigId()) {
-            throw new ClientException('Can\'t fetch component configuration: job has no configId set');
-        }
-
-        try {
-            return $this->componentConfiguration = $this->getComponentsApiClient()->getConfiguration(
-                $this->getComponentId(),
-                $this->getConfigId(),
-            );
-        } catch (StorageApiClientException $e) {
-            throw new ClientException('Failed to fetch component configuration: '.$e->getMessage(), 0, $e);
-        }
+        return $this->componentConfiguration = JobConfigurationResolver::resolveJobConfiguration(
+            $this,
+            $this->getComponentsApiClient(),
+        );
     }
 
     private function getComponentsApiClient(): ComponentsApiClient
diff --git a/src/JobFactory/JobConfigurationResolver.php b/src/JobFactory/JobConfigurationResolver.php
@@ -0,0 +1,28 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Keboola\JobQueueInternalClient\JobFactory;
+
+use Keboola\JobQueueInternalClient\Exception\ClientException;
+use Keboola\StorageApi\ClientException as StorageApiClientException;
+use Keboola\StorageApi\Components as ComponentsApiClient;
+
+class JobConfigurationResolver
+{
+    public static function resolveJobConfiguration(JobInterface $job, ComponentsApiClient $componentsApiClient): array
+    {
+        if (!$job->getConfigId()) {
+            throw new ClientException('Can\'t fetch component configuration: job has no configId set');
+        }
+
+        try {
+            return $componentsApiClient->getConfiguration(
+                $job->getComponentId(),
+                $job->getConfigId(),
+            );
+        } catch (StorageApiClientException $e) {
+            throw new ClientException('Failed to fetch component configuration: '.$e->getMessage(), 0, $e);
+        }
+    }
+}
diff --git a/src/Orchestration/OrchestrationJobMatcher.php b/src/Orchestration/OrchestrationJobMatcher.php
@@ -5,26 +5,34 @@
 namespace Keboola\JobQueueInternalClient\Orchestration;
 
 use Keboola\JobQueueInternalClient\Client;
-use Keboola\JobQueueInternalClient\Exception\ClientException;
 use Keboola\JobQueueInternalClient\Exception\OrchestrationJobMatcherValidationException;
 use Keboola\JobQueueInternalClient\JobFactory;
-use Keboola\JobQueueInternalClient\JobFactory\Job;
 use Keboola\JobQueueInternalClient\JobFactory\JobInterface;
 use Keboola\JobQueueInternalClient\JobListOptions;
+use Keboola\StorageApi\Components;
+use Keboola\StorageApiBranch\Factory\ClientOptions;
+use Keboola\StorageApiBranch\Factory\StorageClientPlainFactory;
+use SensitiveParameter;
 
 // https://keboola.atlassian.net/wiki/spaces/ENGG/pages/3074195457/DRAFT+RFC-2023-011+-+Rerun+orchestration#Pair-Jobs-and-Tasks
 class OrchestrationJobMatcher
 {
     public function __construct(
         private readonly Client $internalClient,
+        private readonly StorageClientPlainFactory $storageClientFactory,
     ) {
     }
 
-    public function matchTaskJobsForOrchestrationJob(string $jobId): OrchestrationJobMatcherResults
-    {
+    public function matchTaskJobsForOrchestrationJob(
+        string $jobId,
+        #[SensitiveParameter] string $token,
+    ): OrchestrationJobMatcherResults {
         $job = $this->internalClient->getJob($jobId);
         $childJobs = $this->getOrchestrationTaskJobs($job);
-        $configuration = $this->getCurrentOrchestrationConfiguration($job);
+        $configuration = $this->getCurrentOrchestrationConfiguration(
+            $job,
+            $this->createComponentsApi($token, $job->getBranchId()),
+        );
         $this->validateInputs($job, $configuration);
         $matchedTasks = [];
         foreach ($configuration['tasks'] as $task) {
@@ -49,7 +57,7 @@ public function matchTaskJobsForOrchestrationJob(string $jobId): OrchestrationJo
             }
         }
         return new OrchestrationJobMatcherResults(
-            $jobId,
+            $job->getId(),
             $job->getConfigId(),
             $matchedTasks,
         );
@@ -66,13 +74,17 @@ private function getOrchestrationTaskJobs(JobInterface $job): array
         );
     }
 
-    private function getCurrentOrchestrationConfiguration(JobInterface $job): array
+    private function getCurrentOrchestrationConfiguration(JobInterface $job, Components $componentsApi): array
     {
         $configuration = $job->getConfigData();
-        if (!$configuration) {
-            $configuration = $job->getComponentConfiguration()['configuration'];
-        };
-        return $configuration;
+        if ($configuration) {
+            return $configuration;
+        }
+
+        return JobFactory\JobConfigurationResolver::resolveJobConfiguration(
+            $job,
+            $componentsApi,
+        )['configuration'];
     }
 
     private function validateInputs(JobInterface $job, array $configuration): void
@@ -101,4 +113,17 @@ private function validateInputs(JobInterface $job, array $configuration): void
             }
         });
     }
+
+    private function createComponentsApi(
+        #[SensitiveParameter] string $token,
+        ?string $branchId,
+    ): Components {
+        return new Components(
+            $this->storageClientFactory->createClientWrapper(
+                (new ClientOptions())
+                    ->setBranchId($branchId)
+                    ->setToken($token),
+            )->getBranchClient(),
+        );
+    }
 }
diff --git a/tests/JobFactory/JobConfigurationResolverTest.php b/tests/JobFactory/JobConfigurationResolverTest.php
@@ -0,0 +1,79 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Keboola\JobQueueInternalClient\Tests\JobFactory;
+
+use Keboola\JobQueueInternalClient\Exception\ClientException;
+use Keboola\JobQueueInternalClient\JobFactory\JobConfigurationResolver;
+use Keboola\JobQueueInternalClient\JobFactory\JobInterface;
+use Keboola\StorageApi\ClientException as StorageApiClientException;
+use Keboola\StorageApi\Components;
+use PHPUnit\Framework\TestCase;
+
+class JobConfigurationResolverTest extends TestCase
+{
+    public function testResolveJobConfiguration(): void
+    {
+        $jobMock = $this->createMock(JobInterface::class);
+        $jobMock->expects(self::exactly(2))
+            ->method('getConfigId')
+            ->willReturn('123')
+        ;
+        $jobMock->expects(self::once())
+            ->method('getComponentId')
+            ->willReturn('keboola.test-component')
+        ;
+
+        $componentsApiClientMock = $this->createMock(Components::class);
+        $componentsApiClientMock->expects(self::once())
+            ->method('getConfiguration')
+            ->with('keboola.test-component', '123')
+            ->willReturn(['id' => '123', 'name' => 'test'])
+        ;
+
+        self::assertSame(
+            ['id' => '123', 'name' => 'test'],
+            JobConfigurationResolver::resolveJobConfiguration($jobMock, $componentsApiClientMock),
+        );
+    }
+
+    public function testResolveJobConfigurationFailsForNoConfigId(): void
+    {
+        $jobMock = $this->createMock(JobInterface::class);
+        $jobMock->expects(self::once())
+            ->method('getConfigId')
+            ->willReturn(null)
+        ;
+
+        $this->expectExceptionMessage('Can\'t fetch component configuration: job has no configId set');
+        JobConfigurationResolver::resolveJobConfiguration(
+            $jobMock,
+            $this->createMock(Components::class),
+        );
+    }
+
+    public function testResolveJobConfigurationHandlesStorageApiErrors(): void
+    {
+        $jobMock = $this->createMock(JobInterface::class);
+        $jobMock->expects(self::exactly(2))
+            ->method('getConfigId')
+            ->willReturn('123')
+        ;
+        $jobMock->expects(self::once())
+            ->method('getComponentId')
+            ->willReturn('keboola.test-component')
+        ;
+
+        $componentsApiClientMock = $this->createMock(Components::class);
+        $componentsApiClientMock->expects(self::once())
+            ->method('getConfiguration')
+            ->with('keboola.test-component', '123')
+            ->willThrowException(new StorageApiClientException('Sample error'))
+        ;
+
+        $this->expectExceptionMessage('Failed to fetch component configuration: Sample error');
+        $this->expectException(ClientException::class);
+        JobConfigurationResolver::resolveJobConfiguration($jobMock, $componentsApiClientMock);
+    }
+}
diff --git a/tests/JobFactory/JobTest.php b/tests/JobFactory/JobTest.php
@@ -653,10 +653,17 @@ public function testGetComponentConfiguration(): void
     public function testGetComponentConfigurationWhenJobHasNoConfigIdSet(): void
     {
         $storageClientFactory = $this->createMock(StorageClientPlainFactory::class);
-        $storageClientFactory->expects(self::never())->method(self::anything());
+        $storageClientFactory->expects(self::once())
+            ->method('createClientWrapper')
+            ->willReturn($this->createMock(ClientWrapper::class))
+        ;
 
         $objectEncryptorMock = $this->createMock(JobObjectEncryptor::class);
-        $objectEncryptorMock->expects(self::never())->method('decrypt');
+        $objectEncryptorMock->expects(self::once())
+            ->method('decrypt')
+            ->with('KBC::ProjectSecure::token')
+            ->willReturn('token-decrypted')
+        ;
 
         $jobData = $this->jobData;
         $jobData['configId'] = null;
diff --git a/tests/Orchestration/OrchestrationJobMatcherTest.php b/tests/Orchestration/OrchestrationJobMatcherTest.php
@@ -5,6 +5,7 @@
 namespace Keboola\JobQueueInternalClient\Tests\Orchestration;
 
 use Generator;
+use Keboola\JobQueueInternalClient\Client;
 use Keboola\JobQueueInternalClient\Exception\OrchestrationJobMatcherValidationException;
 use Keboola\JobQueueInternalClient\JobFactory;
 use Keboola\JobQueueInternalClient\Orchestration\OrchestrationJobMatcher;
@@ -14,6 +15,10 @@
 use Keboola\StorageApi\Client as StorageClient;
 use Keboola\StorageApi\Components;
 use Keboola\StorageApi\Options\Components\Configuration;
+use Keboola\StorageApi\Options\TokenCreateOptions;
+use Keboola\StorageApi\Tokens;
+use Keboola\StorageApiBranch\Factory\ClientOptions;
+use Keboola\StorageApiBranch\Factory\StorageClientPlainFactory;
 
 class OrchestrationJobMatcherTest extends BaseClientFunctionalTest
 {
@@ -96,9 +101,20 @@ private function getOrchestrationConfiguration(): array
     private function createOrchestrationLikeJobs(array $configurationData, array $createOnlyTasks): array
     {
         $storageClient = new StorageClient([
-            'token' => (string) getenv('TEST_STORAGE_API_TOKEN'),
-            'url' => (string) getenv('TEST_STORAGE_API_URL'),
+            'token' => self::getRequiredEnv('TEST_STORAGE_API_TOKEN_MASTER'),
+            'url' => self::getRequiredEnv('TEST_STORAGE_API_URL'),
         ]);
+
+        // create token for job
+        $tokenOptions = (new TokenCreateOptions())
+            ->setDescription(__CLASS__)
+            ->setCanManageBuckets(true) // access to all components :)
+            ->setExpiresIn(60 * 5)
+        ;
+
+        $tokens = new Tokens($storageClient);
+        $token = $tokens->createToken($tokenOptions); // new token, created specially for this job
+
         $queueClient = $this->getClient();
         $componentsApi = new Components($storageClient);
         $configuration = new Configuration();
@@ -107,9 +123,10 @@ private function createOrchestrationLikeJobs(array $configurationData, array $cr
         $configuration->setComponentId(JobFactory::ORCHESTRATOR_COMPONENT);
         $this->componentId = JobFactory::ORCHESTRATOR_COMPONENT;
         $this->configurationId = $componentsApi->addConfiguration($configuration)['id'];
+
         $orchestrationJob = $this->getNewJobFactory()->createNewJob(
             [
-                '#tokenString' => getenv('TEST_STORAGE_API_TOKEN'),
+                '#tokenString' => $token['token'],
                 'configData' => [],
                 'componentId' => JobFactory::ORCHESTRATOR_COMPONENT,
                 'configId' => $this->configurationId,
@@ -133,7 +150,7 @@ private function createOrchestrationLikeJobs(array $configurationData, array $cr
             ];
             $phaseJob = $queueClient->createJob($this->getNewJobFactory()->createNewJob(
                 [
-                    '#tokenString' => getenv('TEST_STORAGE_API_TOKEN'),
+                    '#tokenString' => $token['token'],
                     'configData' => $configData,
                     'componentId' => JobFactory::ORCHESTRATOR_COMPONENT,
                     'configId' => $this->configurationId,
@@ -151,7 +168,7 @@ private function createOrchestrationLikeJobs(array $configurationData, array $cr
             }
             $jobIds[] = $queueClient->createJob($this->getNewJobFactory()->createNewJob(
                 [
-                    '#tokenString' => getenv('TEST_STORAGE_API_TOKEN'),
+                    '#tokenString' => $token['token'],
                     'componentId' => $task['task']['componentId'],
                     'configData' => $task['task']['configData'],
                     'mode' => $task['task']['mode'],
@@ -162,6 +179,8 @@ private function createOrchestrationLikeJobs(array $configurationData, array $cr
             ))->getId();
         }
 
+        $tokens->dropToken($token['id']); // drob token
+
         return [
             'orchestrationJobId' => $orchestrationJob->getId(),
             'orchestrationConfigurationId' => $this->configurationId,
@@ -178,8 +197,11 @@ public function testMatcherFull(): void
             'jobIds' => $jobIds,
         ] = $this->createOrchestrationLikeJobs($this->getOrchestrationConfiguration(), []);
 
-        $matcher = new OrchestrationJobMatcher($client);
-        $results = $matcher->matchTaskJobsForOrchestrationJob($orchestrationJobId);
+        $matcher = new OrchestrationJobMatcher($client, $this->createStorageClientPlainFactory());
+        $results = $matcher->matchTaskJobsForOrchestrationJob(
+            $orchestrationJobId,
+            self::getRequiredEnv('TEST_STORAGE_API_TOKEN'),
+        );
         self::assertEquals(
             new OrchestrationJobMatcherResults(
                 $orchestrationJobId,
@@ -224,8 +246,11 @@ public function testMatcherPartial(): void
             'jobIds' => $jobIds,
         ] = $this->createOrchestrationLikeJobs($this->getOrchestrationConfiguration(), ['92543']);
 
-        $matcher = new OrchestrationJobMatcher($client);
-        $results = $matcher->matchTaskJobsForOrchestrationJob($orchestrationJobId);
+        $matcher = new OrchestrationJobMatcher($client, $this->createStorageClientPlainFactory());
+        $results = $matcher->matchTaskJobsForOrchestrationJob(
+            $orchestrationJobId,
+            self::getRequiredEnv('TEST_STORAGE_API_TOKEN'),
+        );
         self::assertEquals(
             new OrchestrationJobMatcherResults(
                 $orchestrationJobId,
@@ -277,8 +302,8 @@ public function testMatcherInvalidConfiguration(
         string $expectedMessage,
     ): void {
         $storageClient = new StorageClient([
-            'token' => (string) getenv('TEST_STORAGE_API_TOKEN'),
-            'url' => (string) getenv('TEST_STORAGE_API_URL'),
+            'token' => self::getRequiredEnv('TEST_STORAGE_API_TOKEN'),
+            'url' => self::getRequiredEnv('TEST_STORAGE_API_URL'),
         ]);
         $queueClient = $this->getClient();
         $componentsApi = new Components($storageClient);
@@ -290,7 +315,7 @@ public function testMatcherInvalidConfiguration(
         $this->configurationId = $componentsApi->addConfiguration($configuration)['id'];
         $orchestrationJob = $this->getNewJobFactory()->createNewJob(
             [
-                '#tokenString' => getenv('TEST_STORAGE_API_TOKEN'),
+                '#tokenString' => self::getRequiredEnv('TEST_STORAGE_API_TOKEN'),
                 'configData' => [],
                     'componentId' => $componentId,
                 'configId' => $this->configurationId,
@@ -300,10 +325,10 @@ public function testMatcherInvalidConfiguration(
             ],
         );
         $orchestrationJobId = $queueClient->createJob($orchestrationJob)->getId();
-        $matcher = new OrchestrationJobMatcher($queueClient);
+        $matcher = new OrchestrationJobMatcher($queueClient, $this->createStorageClientPlainFactory());
         $this->expectException(OrchestrationJobMatcherValidationException::class);
         $this->expectExceptionMessageMatches($expectedMessage);
-        $matcher->matchTaskJobsForOrchestrationJob($orchestrationJobId);
+        $matcher->matchTaskJobsForOrchestrationJob($orchestrationJobId, self::getRequiredEnv('TEST_STORAGE_API_TOKEN'));
     }
 
     public function invalidConfigurationProvider(): Generator
@@ -340,4 +365,11 @@ public function invalidConfigurationProvider(): Generator
             'expectedMessage' => '#Task does not have an id\. \({"name":"foo","phase":1,"task":{"componentId":"keboola.ex-db-snowflake","configData":\[\],"mode":"run"}}\)#',
         ];
     }
+
+    private function createStorageClientPlainFactory(): StorageClientPlainFactory
+    {
+        return new StorageClientPlainFactory(new ClientOptions(
+            self::getRequiredEnv('TEST_STORAGE_API_URL'),
+        ));
+    }
 }
