@Test public void testTelemetryErrorOutput() throws Exception { ValueProvider<String> schemasLocation = pipeline.newProvider("schemas.tar.gz"); ValueProvider<String> metadataLocation = pipeline .newProvider(Resources.getResource("account-ecosystem/metadata-local.json").getPath()); ValueProvider<Boolean> kmsEnabled = pipeline.newProvider(false); String userId = "2ef67665ec7369ba0fab7f802a5e1e04"; String anonId = encryptWithTestPublicKey(userId); List<String> prevUserIds = ImmutableList.of("3ef67665ec7369ba0fab7f802a5e1e04", "4ef67665ec7369ba0fab7f802a5e1e04"); List<String> prevAnonIds = prevUserIds.stream().map(x -> { try { return encryptWithTestPublicKey(x); } catch (Exception e) { throw new RuntimeException(e); } }).collect(Collectors.toList()); List<String> input = ImmutableList.of("{}", "{\"version\":\"4\",\"payload\":{}}", "{\"version\":\"4\"" + ",\"payload\":{\"ecosystemClientId\":\"foo\",\"ecosystemDeviceId\":\"bar\"}}", Json.asString(ImmutableMap.of("version", "4", "payload", ImmutableMap.builder().put("ecosystemAnonId", anonId) .put("ecosystemClientId", "3ed15efab7e94757bf9e9ef5e844ada2") .put("ecosystemDeviceId", "7ab4e373ce434b848a9d0946e388fee9") .put("previous_ecosystem_anon_ids", prevAnonIds).build()))); List<String> expectedOutput = ImmutableList.of("{\"version\":\"4\"" + ",\"payload\":{\"ecosystemClientId\":\"foo\",\"ecosystemDeviceId\":\"bar\"}}"); List<String> expectedError = ImmutableList.of("{}", "{\"version\":\"4\",\"payload\":{}}", "{\"payload\":{\"ecosystemAnonId\":\"eyJr<435 characters redacted>\"" + ",\"ecosystemClientId\":\"3ed15efab7e94757bf9e9ef5e844ada2\"" + ",\"ecosystemDeviceId\":\"7ab4e373ce434b848a9d0946e388fee9\"" + ",\"previous_ecosystem_anon_ids\":[\"eyJr<435 characters redacted>\"" + ",\"eyJr<435 characters redacted>\"]},\"version\":\"4\"}"); Result<PCollection<PubsubMessage>, PubsubMessage> result = pipeline.apply(Create.of(input)) .apply(InputFileFormat.text.decode()) .apply("AddAttributes", MapElements.into(TypeDescriptor.of(PubsubMessage.class)) .via(element -> new PubsubMessage(element.getPayload(), ImmutableMap.of(Attribute.DOCUMENT_NAMESPACE, "telemetry", Attribute.DOCUMENT_TYPE, "account-ecosystem")))) .apply(DecryptAetIdentifiers.of(schemasLocation, metadataLocation, kmsEnabled)); PAssert.that(result.output().apply("EncodeOutput", OutputFileFormat.text.encode())) .containsInAnyOrder(expectedOutput); PAssert.that(result.failures().apply("Sanitize", SanitizeJsonPayload.of()).apply("EncodeErrors", OutputFileFormat.text.encode())).containsInAnyOrder(expectedError); pipeline.run(); }
public static DecryptAetIdentifiers of(ValueProvider<String> schemasLocation, ValueProvider<String> metadataLocation, ValueProvider<Boolean> kmsEnabled) { return new DecryptAetIdentifiers(schemasLocation, metadataLocation, kmsEnabled); }
DecryptAetIdentifiers extends
    PTransform<PCollection<PubsubMessage>, Result<PCollection<PubsubMessage>, PubsubMessage>> { public static DecryptAetIdentifiers of(ValueProvider<String> schemasLocation, ValueProvider<String> metadataLocation, ValueProvider<Boolean> kmsEnabled) { return new DecryptAetIdentifiers(schemasLocation, metadataLocation, kmsEnabled); } }
DecryptAetIdentifiers extends
    PTransform<PCollection<PubsubMessage>, Result<PCollection<PubsubMessage>, PubsubMessage>> { public static DecryptAetIdentifiers of(ValueProvider<String> schemasLocation, ValueProvider<String> metadataLocation, ValueProvider<Boolean> kmsEnabled) { return new DecryptAetIdentifiers(schemasLocation, metadataLocation, kmsEnabled); } private  DecryptAetIdentifiers(ValueProvider<String> schemasLocation,
      ValueProvider<String> metadataLocation, ValueProvider<Boolean> kmsEnabled); }
DecryptAetIdentifiers extends
    PTransform<PCollection<PubsubMessage>, Result<PCollection<PubsubMessage>, PubsubMessage>> { public static DecryptAetIdentifiers of(ValueProvider<String> schemasLocation, ValueProvider<String> metadataLocation, ValueProvider<Boolean> kmsEnabled) { return new DecryptAetIdentifiers(schemasLocation, metadataLocation, kmsEnabled); } private  DecryptAetIdentifiers(ValueProvider<String> schemasLocation,
      ValueProvider<String> metadataLocation, ValueProvider<Boolean> kmsEnabled); static DecryptAetIdentifiers of(ValueProvider<String> schemasLocation,
      ValueProvider<String> metadataLocation, ValueProvider<Boolean> kmsEnabled); @Override Result<PCollection<PubsubMessage>, PubsubMessage> expand(
      PCollection<PubsubMessage> messages); static JsonNode decrypt(KeyStore keyStore, JsonNode anonIdNode); }
DecryptAetIdentifiers extends
    PTransform<PCollection<PubsubMessage>, Result<PCollection<PubsubMessage>, PubsubMessage>> { public static DecryptAetIdentifiers of(ValueProvider<String> schemasLocation, ValueProvider<String> metadataLocation, ValueProvider<Boolean> kmsEnabled) { return new DecryptAetIdentifiers(schemasLocation, metadataLocation, kmsEnabled); } private  DecryptAetIdentifiers(ValueProvider<String> schemasLocation,
      ValueProvider<String> metadataLocation, ValueProvider<Boolean> kmsEnabled); static DecryptAetIdentifiers of(ValueProvider<String> schemasLocation,
      ValueProvider<String> metadataLocation, ValueProvider<Boolean> kmsEnabled); @Override Result<PCollection<PubsubMessage>, PubsubMessage> expand(
      PCollection<PubsubMessage> messages); static JsonNode decrypt(KeyStore keyStore, JsonNode anonIdNode); static final String ACCOUNT_ECOSYSTEM; }