Inject CloudTasksUtil to AsyncTaskEnqueuer (#1522)

* Inject CloudTasksUtil to AsyncTasksEnqueuer

* Rebase

* Remove QUEUE_ASYNC_DELETE from AsyncTasksEnqueuer

* Refactor create() 

* Remove AppEngineServiceUtil depdendency from AsyncTaskEnqueuer

core/src/main/java/google/registry/batch/AsyncTaskEnqueuer.java

@@ -22,14 +22,17 @@ import com.google.appengine.api.taskqueue.TaskOptions;
 import com.google.appengine.api.taskqueue.TaskOptions.Method;
 import com.google.appengine.api.taskqueue.TransientFailureException;
 import com.google.common.base.Joiner;
+import com.google.common.collect.ArrayListMultimap;
 import com.google.common.collect.ImmutableSortedSet;
+import com.google.common.collect.Multimap;
 import com.google.common.flogger.FluentLogger;
 import google.registry.config.RegistryConfig.Config;
 import google.registry.model.EppResource;
 import google.registry.model.eppcommon.Trid;
 import google.registry.model.host.HostResource;
 import google.registry.persistence.VKey;
-import google.registry.util.AppEngineServiceUtils;
+import google.registry.request.Action.Service;
+import google.registry.util.CloudTasksUtils;
 import google.registry.util.Retrier;
 import javax.inject.Inject;
 import javax.inject.Named;
@@ -58,25 +61,23 @@ public final class AsyncTaskEnqueuer {
   private static final Duration MAX_ASYNC_ETA = Duration.standardDays(30);
 
   private final Duration asyncDeleteDelay;
-  private final Queue asyncActionsPushQueue;
   private final Queue asyncDeletePullQueue;
   private final Queue asyncDnsRefreshPullQueue;
-  private final AppEngineServiceUtils appEngineServiceUtils;
   private final Retrier retrier;
 
+  private CloudTasksUtils cloudTasksUtils;
+
   @Inject
   public AsyncTaskEnqueuer(
-      @Named(QUEUE_ASYNC_ACTIONS) Queue asyncActionsPushQueue,
       @Named(QUEUE_ASYNC_DELETE) Queue asyncDeletePullQueue,
       @Named(QUEUE_ASYNC_HOST_RENAME) Queue asyncDnsRefreshPullQueue,
       @Config("asyncDeleteFlowMapreduceDelay") Duration asyncDeleteDelay,
-      AppEngineServiceUtils appEngineServiceUtils,
+      CloudTasksUtils cloudTasksUtils,
       Retrier retrier) {
-    this.asyncActionsPushQueue = asyncActionsPushQueue;
     this.asyncDeletePullQueue = asyncDeletePullQueue;
     this.asyncDnsRefreshPullQueue = asyncDnsRefreshPullQueue;
     this.asyncDeleteDelay = asyncDeleteDelay;
-    this.appEngineServiceUtils = appEngineServiceUtils;
+    this.cloudTasksUtils = cloudTasksUtils;
     this.retrier = retrier;
   }
 
@@ -102,19 +103,17 @@ public final class AsyncTaskEnqueuer {
           entityKey, firstResave, MAX_ASYNC_ETA);
       return;
     }
-    logger.atInfo().log("Enqueuing async re-save of %s to run at %s.", entityKey, whenToResave);
-    String backendHostname = appEngineServiceUtils.getServiceHostname("backend");
-    TaskOptions task =
-        TaskOptions.Builder.withUrl(ResaveEntityAction.PATH)
-            .method(Method.POST)
-            .header("Host", backendHostname)
-            .countdownMillis(etaDuration.getMillis())
-            .param(PARAM_RESOURCE_KEY, entityKey.stringify())
-            .param(PARAM_REQUESTED_TIME, now.toString());
+    Multimap<String, String> params = ArrayListMultimap.create();
+    params.put(PARAM_RESOURCE_KEY, entityKey.stringify());
+    params.put(PARAM_REQUESTED_TIME, now.toString());
     if (whenToResave.size() > 1) {
-      task.param(PARAM_RESAVE_TIMES, Joiner.on(',').join(whenToResave.tailSet(firstResave, false)));
+      params.put(PARAM_RESAVE_TIMES, Joiner.on(',').join(whenToResave.tailSet(firstResave, false)));
     }
-    addTaskToQueueWithRetry(asyncActionsPushQueue, task);
+    logger.atInfo().log("Enqueuing async re-save of %s to run at %s.", entityKey, whenToResave);
+    cloudTasksUtils.enqueue(
+        QUEUE_ASYNC_ACTIONS,
+        cloudTasksUtils.createPostTaskWithDelay(
+            ResaveEntityAction.PATH, Service.BACKEND.toString(), params, etaDuration));
   }
 
   /** Enqueues a task to asynchronously delete a contact or host, by key. */

core/src/test/java/google/registry/batch/AsyncTaskEnqueuerTest.java

@@ -22,23 +22,20 @@ import static google.registry.batch.AsyncTaskEnqueuer.QUEUE_ASYNC_ACTIONS;
 import static google.registry.batch.AsyncTaskEnqueuer.QUEUE_ASYNC_DELETE;
 import static google.registry.batch.AsyncTaskEnqueuer.QUEUE_ASYNC_HOST_RENAME;
 import static google.registry.testing.DatabaseHelper.persistActiveContact;
-import static google.registry.testing.TaskQueueHelper.assertNoTasksEnqueued;
-import static google.registry.testing.TaskQueueHelper.assertTasksEnqueued;
 import static google.registry.testing.TestLogHandlerUtils.assertLogMessage;
-import static org.joda.time.Duration.standardDays;
-import static org.joda.time.Duration.standardHours;
 import static org.joda.time.Duration.standardSeconds;
-import static org.mockito.Mockito.when;
 
+import com.google.cloud.tasks.v2.HttpMethod;
 import com.google.common.collect.ImmutableSortedSet;
 import google.registry.model.contact.ContactResource;
 import google.registry.testing.AppEngineExtension;
+import google.registry.testing.CloudTasksHelper;
+import google.registry.testing.CloudTasksHelper.TaskMatcher;
 import google.registry.testing.FakeClock;
 import google.registry.testing.FakeSleeper;
 import google.registry.testing.InjectExtension;
-import google.registry.testing.TaskQueueHelper.TaskMatcher;
-import google.registry.util.AppEngineServiceUtils;
 import google.registry.util.CapturingLogHandler;
+import google.registry.util.CloudTasksUtils;
 import google.registry.util.JdkLoggerConfig;
 import google.registry.util.Retrier;
 import java.util.logging.Level;
@@ -48,7 +45,6 @@ import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.junit.jupiter.api.extension.RegisterExtension;
-import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
 import org.mockito.junit.jupiter.MockitoSettings;
 import org.mockito.quality.Strictness;
@@ -63,27 +59,25 @@ public class AsyncTaskEnqueuerTest {
 
   @RegisterExtension public final InjectExtension inject = new InjectExtension();
 
-  @Mock private AppEngineServiceUtils appEngineServiceUtils;
-
   private AsyncTaskEnqueuer asyncTaskEnqueuer;
   private final CapturingLogHandler logHandler = new CapturingLogHandler();
   private final FakeClock clock = new FakeClock(DateTime.parse("2015-05-18T12:34:56Z"));
+  private CloudTasksHelper cloudTasksHelper = new CloudTasksHelper(clock);
 
   @BeforeEach
   void beforeEach() {
     JdkLoggerConfig.getConfig(AsyncTaskEnqueuer.class).addHandler(logHandler);
-    when(appEngineServiceUtils.getServiceHostname("backend")).thenReturn("backend.hostname.fake");
-    asyncTaskEnqueuer = createForTesting(appEngineServiceUtils, clock, standardSeconds(90));
+    asyncTaskEnqueuer =
+        createForTesting(cloudTasksHelper.getTestCloudTasksUtils(), clock, standardSeconds(90));
   }
 
   public static AsyncTaskEnqueuer createForTesting(
-      AppEngineServiceUtils appEngineServiceUtils, FakeClock clock, Duration asyncDeleteDelay) {
+      CloudTasksUtils cloudTasksUtils, FakeClock clock, Duration asyncDeleteDelay) {
     return new AsyncTaskEnqueuer(
-        getQueue(QUEUE_ASYNC_ACTIONS),
         getQueue(QUEUE_ASYNC_DELETE),
         getQueue(QUEUE_ASYNC_HOST_RENAME),
         asyncDeleteDelay,
-        appEngineServiceUtils,
+        cloudTasksUtils,
         new Retrier(new FakeSleeper(clock), 1));
   }
 
@@ -92,18 +86,16 @@ public class AsyncTaskEnqueuerTest {
     ContactResource contact = persistActiveContact("jd23456");
     asyncTaskEnqueuer.enqueueAsyncResave(
         contact.createVKey(), clock.nowUtc(), clock.nowUtc().plusDays(5));
-    assertTasksEnqueued(
+    cloudTasksHelper.assertTasksEnqueued(
         QUEUE_ASYNC_ACTIONS,
-        new TaskMatcher()
+        new CloudTasksHelper.TaskMatcher()
             .url(ResaveEntityAction.PATH)
-            .method("POST")
-            .header("Host", "backend.hostname.fake")
+            .method(HttpMethod.POST)
+            .service("backend")
             .header("content-type", "application/x-www-form-urlencoded")
             .param(PARAM_RESOURCE_KEY, contact.createVKey().stringify())
             .param(PARAM_REQUESTED_TIME, clock.nowUtc().toString())
-            .etaDelta(
-                standardDays(5).minus(standardSeconds(30)),
-                standardDays(5).plus(standardSeconds(30))));
+            .scheduleTime(clock.nowUtc().plus(Duration.standardDays(5))));
   }
 
   @Test
@@ -114,19 +106,17 @@ public class AsyncTaskEnqueuerTest {
         contact.createVKey(),
         now,
         ImmutableSortedSet.of(now.plusHours(24), now.plusHours(50), now.plusHours(75)));
-    assertTasksEnqueued(
+    cloudTasksHelper.assertTasksEnqueued(
         QUEUE_ASYNC_ACTIONS,
         new TaskMatcher()
             .url(ResaveEntityAction.PATH)
-            .method("POST")
-            .header("Host", "backend.hostname.fake")
+            .method(HttpMethod.POST)
+            .service("backend")
             .header("content-type", "application/x-www-form-urlencoded")
             .param(PARAM_RESOURCE_KEY, contact.createVKey().stringify())
             .param(PARAM_REQUESTED_TIME, now.toString())
             .param(PARAM_RESAVE_TIMES, "2015-05-20T14:34:56.000Z,2015-05-21T15:34:56.000Z")
-            .etaDelta(
-                standardHours(24).minus(standardSeconds(30)),
-                standardHours(24).plus(standardSeconds(30))));
+            .scheduleTime(clock.nowUtc().plus(Duration.standardHours(24))));
   }
 
   @MockitoSettings(strictness = Strictness.LENIENT)
@@ -135,7 +125,7 @@ public class AsyncTaskEnqueuerTest {
     ContactResource contact = persistActiveContact("jd23456");
     asyncTaskEnqueuer.enqueueAsyncResave(
         contact.createVKey(), clock.nowUtc(), clock.nowUtc().plusDays(31));
-    assertNoTasksEnqueued(QUEUE_ASYNC_ACTIONS);
+    cloudTasksHelper.assertNoTasksEnqueued(QUEUE_ASYNC_ACTIONS);
     assertLogMessage(logHandler, Level.INFO, "Ignoring async re-save");
   }
 }

core/src/test/java/google/registry/batch/DeleteContactsAndHostsActionTest.java

@@ -91,13 +91,13 @@ import google.registry.model.transfer.ContactTransferData;
 import google.registry.model.transfer.TransferData;
 import google.registry.model.transfer.TransferResponse;
 import google.registry.model.transfer.TransferStatus;
+import google.registry.testing.CloudTasksHelper;
 import google.registry.testing.FakeClock;
 import google.registry.testing.FakeResponse;
 import google.registry.testing.FakeSleeper;
 import google.registry.testing.InjectExtension;
 import google.registry.testing.TaskQueueHelper.TaskMatcher;
 import google.registry.testing.mapreduce.MapreduceTestCase;
-import google.registry.util.AppEngineServiceUtils;
 import google.registry.util.RequestStatusChecker;
 import google.registry.util.Retrier;
 import google.registry.util.Sleeper;
@@ -148,7 +148,7 @@ public class DeleteContactsAndHostsActionTest
     inject.setStaticField(Ofy.class, "clock", clock);
     enqueuer =
         AsyncTaskEnqueuerTest.createForTesting(
-            mock(AppEngineServiceUtils.class), clock, Duration.ZERO);
+            new CloudTasksHelper(clock).getTestCloudTasksUtils(), clock, Duration.ZERO);
     AsyncTaskMetrics asyncTaskMetricsMock = mock(AsyncTaskMetrics.class);
     action = new DeleteContactsAndHostsAction();
     action.asyncTaskMetrics = asyncTaskMetricsMock;

core/src/test/java/google/registry/batch/DeleteExpiredDomainsActionTest.java

@@ -39,7 +39,6 @@ import google.registry.model.domain.DomainHistory;
 import google.registry.model.ofy.Ofy;
 import google.registry.model.poll.PollMessage;
 import google.registry.model.reporting.HistoryEntry;
-import google.registry.monitoring.whitebox.EppMetric;
 import google.registry.persistence.transaction.QueryComposer.Comparator;
 import google.registry.testing.AppEngineExtension;
 import google.registry.testing.DualDatabaseTest;
@@ -78,8 +77,7 @@ class DeleteExpiredDomainsActionTest {
     createTld("tld");
     EppController eppController =
         DaggerEppTestComponent.builder()
-            .fakesAndMocksModule(
-                FakesAndMocksModule.create(clock, EppMetric.builderForRequest(clock)))
+            .fakesAndMocksModule(FakesAndMocksModule.create(clock))
             .build()
             .startRequest()
             .eppController();

core/src/test/java/google/registry/batch/RefreshDnsOnHostRenameActionTest.java

@@ -49,6 +49,7 @@ import google.registry.batch.RefreshDnsOnHostRenameAction.RefreshDnsOnHostRename
 import google.registry.dns.DnsQueue;
 import google.registry.model.host.HostResource;
 import google.registry.model.server.Lock;
+import google.registry.testing.CloudTasksHelper;
 import google.registry.testing.DualDatabaseTest;
 import google.registry.testing.FakeClock;
 import google.registry.testing.FakeResponse;
@@ -59,7 +60,6 @@ import google.registry.testing.TestOfyAndSql;
 import google.registry.testing.TestOfyOnly;
 import google.registry.testing.TestSqlOnly;
 import google.registry.testing.mapreduce.MapreduceTestCase;
-import google.registry.util.AppEngineServiceUtils;
 import google.registry.util.RequestStatusChecker;
 import google.registry.util.Retrier;
 import google.registry.util.Sleeper;
@@ -89,7 +89,7 @@ public class RefreshDnsOnHostRenameActionTest
     createTld("tld");
     enqueuer =
         AsyncTaskEnqueuerTest.createForTesting(
-            mock(AppEngineServiceUtils.class), clock, Duration.ZERO);
+            new CloudTasksHelper(clock).getTestCloudTasksUtils(), clock, Duration.ZERO);
     AsyncTaskMetrics asyncTaskMetricsMock = mock(AsyncTaskMetrics.class);
     action = new RefreshDnsOnHostRenameAction();
     action.asyncTaskMetrics = asyncTaskMetricsMock;

core/src/test/java/google/registry/batch/ResaveEntityActionTest.java

@@ -25,12 +25,10 @@ import static google.registry.testing.DatabaseHelper.persistActiveContact;
 import static google.registry.testing.DatabaseHelper.persistDomainWithDependentResources;
 import static google.registry.testing.DatabaseHelper.persistDomainWithPendingTransfer;
 import static google.registry.testing.DatabaseHelper.persistResource;
-import static google.registry.testing.TaskQueueHelper.assertTasksEnqueued;
 import static org.joda.time.Duration.standardDays;
-import static org.joda.time.Duration.standardSeconds;
 import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
 
+import com.google.cloud.tasks.v2.HttpMethod;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.ImmutableSortedSet;
 import google.registry.model.domain.DomainBase;
@@ -40,10 +38,11 @@ import google.registry.model.eppcommon.StatusValue;
 import google.registry.model.ofy.Ofy;
 import google.registry.request.Response;
 import google.registry.testing.AppEngineExtension;
+import google.registry.testing.CloudTasksHelper;
+import google.registry.testing.CloudTasksHelper.TaskMatcher;
 import google.registry.testing.DualDatabaseTest;
 import google.registry.testing.FakeClock;
 import google.registry.testing.InjectExtension;
-import google.registry.testing.TaskQueueHelper.TaskMatcher;
 import google.registry.testing.TestOfyAndSql;
 import google.registry.util.AppEngineServiceUtils;
 import org.joda.time.DateTime;
@@ -71,13 +70,14 @@ public class ResaveEntityActionTest {
   @Mock private Response response;
   private final FakeClock clock = new FakeClock(DateTime.parse("2016-02-11T10:00:00Z"));
   private AsyncTaskEnqueuer asyncTaskEnqueuer;
+  private CloudTasksHelper cloudTasksHelper = new CloudTasksHelper(clock);
 
   @BeforeEach
   void beforeEach() {
     inject.setStaticField(Ofy.class, "clock", clock);
-    when(appEngineServiceUtils.getServiceHostname("backend")).thenReturn("backend.hostname.fake");
     asyncTaskEnqueuer =
-        AsyncTaskEnqueuerTest.createForTesting(appEngineServiceUtils, clock, Duration.ZERO);
+        AsyncTaskEnqueuerTest.createForTesting(
+            cloudTasksHelper.getTestCloudTasksUtils(), clock, Duration.ZERO);
     createTld("tld");
   }
 
@@ -143,17 +143,15 @@ public class ResaveEntityActionTest {
     DomainBase resavedDomain = loadByEntity(domain);
     assertThat(resavedDomain.getGracePeriods()).isEmpty();
 
-    assertTasksEnqueued(
+    cloudTasksHelper.assertTasksEnqueued(
         QUEUE_ASYNC_ACTIONS,
         new TaskMatcher()
             .url(ResaveEntityAction.PATH)
-            .method("POST")
-            .header("Host", "backend.hostname.fake")
+            .method(HttpMethod.POST)
+            .service("backend")
             .header("content-type", "application/x-www-form-urlencoded")
             .param(PARAM_RESOURCE_KEY, resavedDomain.createVKey().stringify())
             .param(PARAM_REQUESTED_TIME, requestedTime.toString())
-            .etaDelta(
-                standardDays(5).minus(standardSeconds(30)),
-                standardDays(5).plus(standardSeconds(30))));
+            .scheduleTime(clock.nowUtc().plus(standardDays(5))));
   }
 }

core/src/test/java/google/registry/flows/EppPointInTimeTest.java

@@ -72,7 +72,7 @@ class EppPointInTimeTest {
     SessionMetadata sessionMetadata = new HttpSessionMetadata(new FakeHttpSession());
     sessionMetadata.setRegistrarId("TheRegistrar");
     DaggerEppTestComponent.builder()
-        .fakesAndMocksModule(FakesAndMocksModule.create(clock, EppMetric.builderForRequest(clock)))
+        .fakesAndMocksModule(FakesAndMocksModule.create(clock))
         .build()
         .startRequest()
         .flowComponentBuilder()

core/src/test/java/google/registry/flows/EppTestCase.java

@@ -224,12 +224,14 @@ public class EppTestCase {
     EppRequestHandler handler = new EppRequestHandler();
     FakeResponse response = new FakeResponse();
     handler.response = response;
-    eppMetricBuilder = EppMetric.builderForRequest(clock);
-    handler.eppController = DaggerEppTestComponent.builder()
-        .fakesAndMocksModule(FakesAndMocksModule.create(clock, eppMetricBuilder))
-        .build()
-        .startRequest()
-        .eppController();
+    FakesAndMocksModule fakesAndMocksModule = FakesAndMocksModule.create(clock);
+    eppMetricBuilder = fakesAndMocksModule.getMetricBuilder();
+    handler.eppController =
+        DaggerEppTestComponent.builder()
+            .fakesAndMocksModule(fakesAndMocksModule)
+            .build()
+            .startRequest()
+            .eppController();
     handler.executeEpp(
         sessionMetadata,
         credentials,

core/src/test/java/google/registry/flows/EppTestComponent.java

@@ -15,8 +15,6 @@
 package google.registry.flows;
 
 import static org.joda.time.Duration.standardSeconds;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
 
 import dagger.Component;
 import dagger.Module;
@@ -33,12 +31,12 @@ import google.registry.flows.domain.DomainFlowTmchUtils;
 import google.registry.monitoring.whitebox.EppMetric;
 import google.registry.request.RequestScope;
 import google.registry.request.lock.LockHandler;
+import google.registry.testing.CloudTasksHelper;
 import google.registry.testing.FakeClock;
 import google.registry.testing.FakeLockHandler;
 import google.registry.testing.FakeSleeper;
 import google.registry.tmch.TmchCertificateAuthority;
 import google.registry.tmch.TmchXmlSignature;
-import google.registry.util.AppEngineServiceUtils;
 import google.registry.util.Clock;
 import google.registry.util.Sleeper;
 import javax.inject.Singleton;
@@ -60,35 +58,32 @@ public interface EppTestComponent {
     private EppMetric.Builder metricBuilder;
     private FakeClock clock;
     private FakeLockHandler lockHandler;
-    private AppEngineServiceUtils appEngineServiceUtils;
     private Sleeper sleeper;
+    private CloudTasksHelper cloudTasksHelper;
 
-    public static FakesAndMocksModule create() {
-      FakeClock clock = new FakeClock();
-      return create(clock, EppMetric.builderForRequest(clock));
+    public CloudTasksHelper getCloudTasksHelper() {
+      return cloudTasksHelper;
     }
 
-    public static FakesAndMocksModule create(FakeClock clock, EppMetric.Builder metricBuilder) {
-      return create(
-          clock,
-          metricBuilder,
-          new TmchXmlSignature(new TmchCertificateAuthority(TmchCaMode.PILOT, clock)));
+    public EppMetric.Builder getMetricBuilder() {
+      return metricBuilder;
     }
 
-    public static FakesAndMocksModule create(
-        FakeClock clock, EppMetric.Builder eppMetricBuilder, TmchXmlSignature tmchXmlSignature) {
+    public static FakesAndMocksModule create(FakeClock clock) {
       FakesAndMocksModule instance = new FakesAndMocksModule();
-      AppEngineServiceUtils appEngineServiceUtils = mock(AppEngineServiceUtils.class);
-      when(appEngineServiceUtils.getServiceHostname("backend")).thenReturn("backend.hostname.fake");
+      CloudTasksHelper cloudTasksHelper = new CloudTasksHelper(clock);
       instance.asyncTaskEnqueuer =
-          AsyncTaskEnqueuerTest.createForTesting(appEngineServiceUtils, clock, standardSeconds(90));
+          AsyncTaskEnqueuerTest.createForTesting(
+              cloudTasksHelper.getTestCloudTasksUtils(), clock, standardSeconds(90));
       instance.clock = clock;
-      instance.domainFlowTmchUtils = new DomainFlowTmchUtils(tmchXmlSignature);
-      instance.sleeper = new FakeSleeper(clock);
+      instance.domainFlowTmchUtils =
+          new DomainFlowTmchUtils(
+              new TmchXmlSignature(new TmchCertificateAuthority(TmchCaMode.PILOT, clock)));
+      instance.sleeper = new FakeSleeper(instance.clock);
       instance.dnsQueue = DnsQueue.create();
-      instance.metricBuilder = eppMetricBuilder;
-      instance.appEngineServiceUtils = appEngineServiceUtils;
+      instance.metricBuilder = EppMetric.builderForRequest(clock);
       instance.lockHandler = new FakeLockHandler(true);
+      instance.cloudTasksHelper = cloudTasksHelper;
       return instance;
     }
 
@@ -127,11 +122,6 @@ public interface EppTestComponent {
       return metricBuilder;
     }
 
-    @Provides
-    AppEngineServiceUtils provideAppEngineServiceUtils() {
-      return appEngineServiceUtils;
-    }
-
     @Provides
     Sleeper provideSleeper() {
       return sleeper;

core/src/test/java/google/registry/flows/FlowTestCase.java

@@ -33,7 +33,6 @@ import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Maps;
 import com.google.common.collect.ObjectArrays;
-import google.registry.config.RegistryConfig.ConfigModule.TmchCaMode;
 import google.registry.flows.EppTestComponent.FakesAndMocksModule;
 import google.registry.flows.picker.FlowPicker;
 import google.registry.model.billing.BillingEvent;
@@ -45,14 +44,13 @@ import google.registry.model.ofy.Ofy;
 import google.registry.model.reporting.HistoryEntryDao;
 import google.registry.monitoring.whitebox.EppMetric;
 import google.registry.testing.AppEngineExtension;
+import google.registry.testing.CloudTasksHelper;
 import google.registry.testing.DatabaseHelper;
 import google.registry.testing.EppLoader;
 import google.registry.testing.FakeClock;
 import google.registry.testing.FakeHttpSession;
 import google.registry.testing.InjectExtension;
 import google.registry.testing.TestDataHelper;
-import google.registry.tmch.TmchCertificateAuthority;
-import google.registry.tmch.TmchXmlSignature;
 import google.registry.util.TypeUtils.TypeInstantiator;
 import google.registry.xml.ValidationMode;
 import java.util.Arrays;
@@ -88,7 +86,7 @@ public abstract class FlowTestCase<F extends Flow> {
   protected FakeClock clock = new FakeClock(DateTime.now(UTC));
   protected TransportCredentials credentials = new PasswordOnlyTransportCredentials();
   protected EppRequestSource eppRequestSource = EppRequestSource.UNIT_TEST;
-  private TmchXmlSignature testTmchXmlSignature = null;
+  protected CloudTasksHelper cloudTasksHelper;
 
   private EppMetric.Builder eppMetricBuilder;
 
@@ -229,13 +227,12 @@ public abstract class FlowTestCase<F extends Flow> {
     // Assert that the xml triggers the flow we expect.
     assertThat(FlowPicker.getFlowClass(eppLoader.getEpp()))
         .isEqualTo(new TypeInstantiator<F>(getClass()){}.getExactType());
+
+    FakesAndMocksModule fakesAndMocksModule = FakesAndMocksModule.create(clock);
+    cloudTasksHelper = fakesAndMocksModule.getCloudTasksHelper();
     // Run the flow.
-    TmchXmlSignature tmchXmlSignature =
-        testTmchXmlSignature != null
-            ? testTmchXmlSignature
-            : new TmchXmlSignature(new TmchCertificateAuthority(tmchCaMode, clock));
     return DaggerEppTestComponent.builder()
-        .fakesAndMocksModule(FakesAndMocksModule.create(clock, eppMetricBuilder, tmchXmlSignature))
+        .fakesAndMocksModule(fakesAndMocksModule)
         .build()
         .startRequest()
         .flowComponentBuilder()
@@ -300,8 +297,6 @@ public abstract class FlowTestCase<F extends Flow> {
     return output;
   }
 
-  private TmchCaMode tmchCaMode = TmchCaMode.PILOT;
-
   public EppOutput dryRunFlowAssertResponse(String xml, String... ignoredPaths) throws Exception {
     List<Object> beforeEntities = DatabaseHelper.loadAllEntities();
     EppOutput output =

core/src/test/java/google/registry/flows/domain/DomainDeleteFlowTest.java

@@ -54,14 +54,13 @@ import static google.registry.testing.DomainBaseSubject.assertAboutDomains;
 import static google.registry.testing.EppExceptionSubject.assertAboutEppExceptions;
 import static google.registry.testing.HistoryEntrySubject.assertAboutHistoryEntries;
 import static google.registry.testing.TaskQueueHelper.assertDnsTasksEnqueued;
-import static google.registry.testing.TaskQueueHelper.assertTasksEnqueued;
 import static google.registry.util.DateTimeUtils.END_OF_TIME;
 import static google.registry.util.DateTimeUtils.START_OF_TIME;
 import static org.joda.money.CurrencyUnit.USD;
 import static org.joda.time.Duration.standardDays;
-import static org.joda.time.Duration.standardSeconds;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 
+import com.google.cloud.tasks.v2.HttpMethod;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
@@ -102,10 +101,10 @@ import google.registry.model.tld.Registry.TldType;
 import google.registry.model.transfer.DomainTransferData;
 import google.registry.model.transfer.TransferResponse;
 import google.registry.model.transfer.TransferStatus;
+import google.registry.testing.CloudTasksHelper.TaskMatcher;
 import google.registry.testing.DatabaseHelper;
 import google.registry.testing.DualDatabaseTest;
 import google.registry.testing.ReplayExtension;
-import google.registry.testing.TaskQueueHelper.TaskMatcher;
 import google.registry.testing.TestOfyAndSql;
 import google.registry.testing.TestOfyOnly;
 import java.util.Map;
@@ -314,17 +313,17 @@ class DomainDeleteFlowTest extends ResourceFlowTestCase<DomainDeleteFlow, Domain
     clock.advanceOneMilli();
     runFlowAssertResponse(loadFile("domain_delete_response_pending.xml"));
     Duration when = standardDays(3);
-    assertTasksEnqueued(
+    cloudTasksHelper.assertTasksEnqueued(
         QUEUE_ASYNC_ACTIONS,
         new TaskMatcher()
             .url(ResaveEntityAction.PATH)
-            .method("POST")
-            .header("Host", "backend.hostname.fake")
+            .method(HttpMethod.POST)
+            .service("backend")
             .header("content-type", "application/x-www-form-urlencoded")
             .param(PARAM_RESOURCE_KEY, domain.createVKey().stringify())
             .param(PARAM_REQUESTED_TIME, clock.nowUtc().toString())
             .param(PARAM_RESAVE_TIMES, clock.nowUtc().plusDays(5).toString())
-            .etaDelta(when.minus(standardSeconds(30)), when.plus(standardSeconds(30))));
+            .scheduleTime(clock.nowUtc().plus(when)));
   }
 
   @TestOfyAndSql

core/src/test/java/google/registry/flows/domain/DomainTransferRequestFlowTest.java

@@ -43,12 +43,11 @@ import static google.registry.testing.DomainBaseSubject.assertAboutDomains;
 import static google.registry.testing.EppExceptionSubject.assertAboutEppExceptions;
 import static google.registry.testing.HistoryEntrySubject.assertAboutHistoryEntries;
 import static google.registry.testing.HostResourceSubject.assertAboutHosts;
-import static google.registry.testing.TaskQueueHelper.assertTasksEnqueued;
 import static google.registry.util.DateTimeUtils.START_OF_TIME;
 import static org.joda.money.CurrencyUnit.USD;
-import static org.joda.time.Duration.standardSeconds;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 
+import com.google.cloud.tasks.v2.HttpMethod;
 import com.google.common.base.Predicates;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
@@ -106,10 +105,10 @@ import google.registry.model.transfer.DomainTransferData;
 import google.registry.model.transfer.TransferResponse;
 import google.registry.model.transfer.TransferStatus;
 import google.registry.persistence.VKey;
+import google.registry.testing.CloudTasksHelper.TaskMatcher;
 import google.registry.testing.DatabaseHelper;
 import google.registry.testing.DualDatabaseTest;
 import google.registry.testing.ReplayExtension;
-import google.registry.testing.TaskQueueHelper.TaskMatcher;
 import google.registry.testing.TestOfyAndSql;
 import google.registry.testing.TestOfyOnly;
 import java.util.Map;
@@ -514,18 +513,16 @@ class DomainTransferRequestFlowTest
     assertPollMessagesEmitted(expectedExpirationTime, implicitTransferTime);
     assertAboutDomainAfterAutomaticTransfer(
         expectedExpirationTime, implicitTransferTime, Period.create(1, Unit.YEARS));
-    assertTasksEnqueued(
+    cloudTasksHelper.assertTasksEnqueued(
         QUEUE_ASYNC_ACTIONS,
         new TaskMatcher()
             .url(ResaveEntityAction.PATH)
-            .method("POST")
-            .header("Host", "backend.hostname.fake")
+            .method(HttpMethod.POST)
+            .service("backend")
             .header("content-type", "application/x-www-form-urlencoded")
             .param(PARAM_RESOURCE_KEY, domain.createVKey().stringify())
             .param(PARAM_REQUESTED_TIME, clock.nowUtc().toString())
-            .etaDelta(
-                registry.getAutomaticTransferLength().minus(standardSeconds(30)),
-                registry.getAutomaticTransferLength().plus(standardSeconds(30))));
+            .scheduleTime(clock.nowUtc().plus(registry.getAutomaticTransferLength())));
   }
 
   private void doSuccessfulTest(