test: cover the heartbeat serialization context

The bug reached a release because heartbeat details were never read back
through a context aware converter.

Author: xepozz

tests/Unit/Router/ContextSignatureConverter.php

@@ -0,0 +1,50 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Temporal\Tests\Unit\Router;
+
+use Temporal\Api\Common\V1\Payload;
+use Temporal\DataConverter\ActivitySerializationContext;
+use Temporal\DataConverter\PayloadConverterInterface;
+use Temporal\DataConverter\SerializationContext;
+use Temporal\DataConverter\SerializationContextAwareInterface;
+use Temporal\DataConverter\Type;
+
+final class ContextSignatureConverter implements PayloadConverterInterface, SerializationContextAwareInterface
+{
+    public const ENCODING = 'test/context-signature';
+    public const NO_CONTEXT = 'none';
+
+    private string $signature = self::NO_CONTEXT;
+
+    public function getEncodingType(): string
+    {
+        return self::ENCODING;
+    }
+
+    public function getSerializationContext(): ?SerializationContext
+    {
+        return null;
+    }
+
+    public function withSerializationContext(?SerializationContext $context): static
+    {
+        $clone = clone $this;
+        $clone->signature = $context instanceof ActivitySerializationContext
+            ? \sprintf('act|%s|%s|%s', $context->namespace, $context->activityType, $context->taskQueue)
+            : self::NO_CONTEXT;
+
+        return $clone;
+    }
+
+    public function toPayload($value): ?Payload
+    {
+        return (new Payload())->setData('')->setMetadata(['encoding' => self::ENCODING]);
+    }
+
+    public function fromPayload(Payload $payload, Type $type): mixed
+    {
+        return $this->signature;
+    }
+}

tests/Unit/Router/HeartbeatDetailsActivity.php

@@ -0,0 +1,20 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Temporal\Tests\Unit\Router;
+
+use Temporal\Activity;
+use Temporal\Activity\ActivityInterface;
+use Temporal\Activity\ActivityMethod;
+use Temporal\DataConverter\Type;
+
+#[ActivityInterface(prefix: 'HeartbeatDetailsActivity.')]
+final class HeartbeatDetailsActivity
+{
+    #[ActivityMethod(name: 'ReadSignature')]
+    public function readSignature(): string
+    {
+        return Activity::getHeartbeatDetails(Type::TYPE_STRING);
+    }
+}

tests/Unit/Router/InvokeActivityHeartbeatContextTestCase.php

@@ -0,0 +1,111 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Temporal\Tests\Unit\Router;
+
+use DateTimeImmutable;
+use Psr\Log\NullLogger;
+use React\Promise\Deferred;
+use Spiral\Attributes\AnnotationReader;
+use Spiral\Attributes\AttributeReader;
+use Spiral\Attributes\Composite\SelectiveReader;
+use Spiral\Attributes\ReaderInterface;
+use Temporal\DataConverter\DataConverter;
+use Temporal\DataConverter\EncodedValues;
+use Temporal\DataConverter\ValuesInterface;
+use Temporal\Exception\ExceptionInterceptorInterface;
+use Temporal\Interceptor\SimplePipelineProvider;
+use Temporal\Internal\Declaration\Reader\ActivityReader;
+use Temporal\Internal\Marshaller\Mapper\AttributeMapperFactory;
+use Temporal\Internal\Marshaller\Marshaller;
+use Temporal\Internal\Queue\QueueInterface;
+use Temporal\Internal\ServiceContainer;
+use Temporal\Internal\Transport\ClientInterface;
+use Temporal\Internal\Transport\Router\InvokeActivity;
+use Temporal\Tests\Unit\AbstractUnit;
+use Temporal\Worker\Environment\EnvironmentInterface;
+use Temporal\Worker\LoopInterface;
+use Temporal\Worker\Transport\Command\Server\ServerRequest;
+use Temporal\Worker\Transport\Command\Server\TickInfo;
+use Temporal\Worker\Transport\RPCConnectionInterface;
+
+final class InvokeActivityHeartbeatContextTestCase extends AbstractUnit
+{
+    private const NAMESPACE = 'test-namespace';
+    private const TASK_QUEUE = 'test-task-queue';
+    private const ACTIVITY = 'HeartbeatDetailsActivity.ReadSignature';
+
+    public function testHeartbeatDetailsAreDecodedWithTheActivityContext(): void
+    {
+        $signature = $this->runActivity();
+
+        $this->assertSame(
+            \sprintf('act|%s|%s|%s', self::NAMESPACE, self::ACTIVITY, self::TASK_QUEUE),
+            $signature,
+        );
+    }
+
+    private function runActivity(): string
+    {
+        $converter = new DataConverter(new ContextSignatureConverter());
+        $services = $this->createServices($converter);
+        $router = new InvokeActivity($services, $this->createMock(RPCConnectionInterface::class), new SimplePipelineProvider());
+
+        $resolver = new Deferred();
+        $router->handle($this->createRequest($converter), [], $resolver);
+
+        $result = null;
+        $resolver->promise()->then(static function (ValuesInterface $values) use (&$result): void {
+            $result = $values->getValue(0);
+        });
+
+        return $result;
+    }
+
+    private function createRequest(DataConverter $converter): ServerRequest
+    {
+        $options = [
+            'name' => self::ACTIVITY,
+            'heartbeatDetails' => 1,
+            'info' => [
+                'TaskToken' => \base64_encode('token'),
+                'ActivityType' => ['Name' => self::ACTIVITY],
+                'ActivityID' => '1',
+                'WorkflowNamespace' => self::NAMESPACE,
+                'TaskQueue' => self::TASK_QUEUE,
+                'Attempt' => 2,
+            ],
+        ];
+
+        return new ServerRequest(
+            'InvokeActivity',
+            new TickInfo(new DateTimeImmutable()),
+            $options,
+            EncodedValues::fromValues(['input', 'heartbeat'], $converter),
+        );
+    }
+
+    private function createServices(DataConverter $converter): ServiceContainer
+    {
+        $services = new ServiceContainer(
+            $this->createMock(LoopInterface::class),
+            $this->createMock(EnvironmentInterface::class),
+            $this->createMock(ClientInterface::class),
+            $this->createMock(ReaderInterface::class),
+            $this->createMock(QueueInterface::class),
+            new Marshaller(new AttributeMapperFactory(new AttributeReader())),
+            $converter,
+            $this->createMock(ExceptionInterceptorInterface::class),
+            new SimplePipelineProvider(),
+            new NullLogger(),
+        );
+
+        $reader = new ActivityReader(new SelectiveReader([new AnnotationReader(), new AttributeReader()]));
+        foreach ($reader->fromClass(HeartbeatDetailsActivity::class) as $proto) {
+            $services->activities->add($proto);
+        }
+
+        return $services;
+    }
+}