#294

Committed 05 Aug 2024 05:36PM UTC coverage: 77.709% (+0.09%) from 77.618%

Build # #294

Build Type

push

github

Committed by

web-flow

Commit Message

Test server Nexus endpoint operator apis (#2162)

* Bump API version to v1.36.0

* Nexus endpoint test server CRUD API implementation

* cleanup

* functional tests

* test operator service external setup

* test environment setup

* test environment setup

* skip functional tests with external server

Run Details

132 of 138 new or added lines in 4 files covered. (95.65%)

4 existing lines in 2 files now uncovered.

19878 of 25580 relevant lines covered (77.71%)

0.78 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

73.33

/temporal-test-server/src/main/java/io/temporal/internal/testservice/TestOperatorService.java

/*
 * Copyright (C) 2022 Temporal Technologies, Inc. All Rights Reserved.
 *
 * Copyright (C) 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
 *
 * Modifications copyright (C) 2017 Uber Technologies, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this material except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package io.temporal.internal.testservice;

import com.google.protobuf.ByteString;
import io.grpc.Status;
import io.grpc.StatusRuntimeException;
import io.grpc.stub.StreamObserver;
import io.temporal.api.enums.v1.IndexedValueType;
import io.temporal.api.nexus.v1.Endpoint;
import io.temporal.api.operatorservice.v1.*;
import java.io.Closeable;
import java.util.List;
import java.util.Map;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * In memory implementation of the Operator Service. To be used for testing purposes only.
 *
 * <p>Do not use directly, instead use {@link io.temporal.testing.TestWorkflowEnvironment}.
 */
final class TestOperatorService extends OperatorServiceGrpc.OperatorServiceImplBase
    implements Closeable {
  private static final Logger log = LoggerFactory.getLogger(TestOperatorService.class);

  private final TestVisibilityStore visibilityStore;
  private final TestNexusEndpointStore nexusEndpointStore;

  public TestOperatorService(
      TestVisibilityStore visibilityStore, TestNexusEndpointStore nexusEndpointStore) {
    this.visibilityStore = visibilityStore;
    this.nexusEndpointStore = nexusEndpointStore;
  }

  @Override
  public void addSearchAttributes(
      AddSearchAttributesRequest request,
      StreamObserver<AddSearchAttributesResponse> responseObserver) {
    try {
      Map<String, IndexedValueType> registeredSearchAttributes =
          visibilityStore.getRegisteredSearchAttributes();
      request.getSearchAttributesMap().keySet().stream()
          .filter(registeredSearchAttributes::containsKey)
          .findFirst()
          .ifPresent(
              sa -> {
                throw Status.ALREADY_EXISTS
                    .withDescription("Search attribute " + sa + " already exists.")
                    .asRuntimeException();
              });
      request.getSearchAttributesMap().forEach(visibilityStore::addSearchAttribute);
      responseObserver.onNext(AddSearchAttributesResponse.newBuilder().build());
      responseObserver.onCompleted();
    } catch (StatusRuntimeException e) {
      handleStatusRuntimeException(e, responseObserver);
    }
  }

  @Override
  public void removeSearchAttributes(
      RemoveSearchAttributesRequest request,
      StreamObserver<RemoveSearchAttributesResponse> responseObserver) {
    try {
      Map<String, IndexedValueType> registeredSearchAttributes =
          visibilityStore.getRegisteredSearchAttributes();
      request.getSearchAttributesList().stream()
          .filter(k -> !registeredSearchAttributes.containsKey(k))
          .findFirst()
          .ifPresent(
              sa -> {
                throw Status.NOT_FOUND
                    .withDescription("Search attribute " + sa + " doesn't exist.")
                    .asRuntimeException();
              });
      request.getSearchAttributesList().forEach(visibilityStore::removeSearchAttribute);
      responseObserver.onNext(RemoveSearchAttributesResponse.newBuilder().build());
      responseObserver.onCompleted();
    } catch (StatusRuntimeException e) {
      handleStatusRuntimeException(e, responseObserver);
    }
  }

  @Override
  public void getNexusEndpoint(
      GetNexusEndpointRequest request, StreamObserver<GetNexusEndpointResponse> responseObserver) {
    try {
      Endpoint endpoint = nexusEndpointStore.getEndpoint(request.getId());
      responseObserver.onNext(GetNexusEndpointResponse.newBuilder().setEndpoint(endpoint).build());
      responseObserver.onCompleted();
    } catch (StatusRuntimeException e) {
      handleStatusRuntimeException(e, responseObserver);
    }
  }

  @Override
  public void createNexusEndpoint(
      CreateNexusEndpointRequest request,
      StreamObserver<CreateNexusEndpointResponse> responseObserver) {
    try {
      Endpoint created = nexusEndpointStore.createEndpoint(request.getSpec());
      responseObserver.onNext(
          CreateNexusEndpointResponse.newBuilder().setEndpoint(created).build());
      responseObserver.onCompleted();
    } catch (StatusRuntimeException e) {
      handleStatusRuntimeException(e, responseObserver);
    }
  }

  @Override
  public void updateNexusEndpoint(
      UpdateNexusEndpointRequest request,
      StreamObserver<UpdateNexusEndpointResponse> responseObserver) {
    try {
      Endpoint updated =
          nexusEndpointStore.updateEndpoint(
              request.getId(), request.getVersion(), request.getSpec());
      responseObserver.onNext(
          UpdateNexusEndpointResponse.newBuilder().setEndpoint(updated).build());
      responseObserver.onCompleted();
    } catch (StatusRuntimeException e) {
      handleStatusRuntimeException(e, responseObserver);
    }
  }

  @Override
  public void deleteNexusEndpoint(
      DeleteNexusEndpointRequest request,
      StreamObserver<DeleteNexusEndpointResponse> responseObserver) {
    try {
      nexusEndpointStore.deleteEndpoint(request.getId(), request.getVersion());
      responseObserver.onNext(DeleteNexusEndpointResponse.newBuilder().build());
      responseObserver.onCompleted();
    } catch (StatusRuntimeException e) {
      handleStatusRuntimeException(e, responseObserver);
    }
  }

  @Override
  public void listNexusEndpoints(
      ListNexusEndpointsRequest request,
      StreamObserver<ListNexusEndpointsResponse> responseObserver) {
    try {
      List<Endpoint> endpoints =
          nexusEndpointStore.listEndpoints(
              request.getPageSize(), request.getNextPageToken().toByteArray(), request.getName());
      ByteString nextPageToken =
          (!endpoints.isEmpty() && endpoints.size() == request.getPageSize())
              ? endpoints.get(endpoints.size() - 1).getIdBytes()
              : ByteString.empty();
      responseObserver.onNext(
          ListNexusEndpointsResponse.newBuilder()
              .addAllEndpoints(endpoints)
              .setNextPageToken(nextPageToken)
              .build());
      responseObserver.onCompleted();
    } catch (StatusRuntimeException e) {
      handleStatusRuntimeException(e, responseObserver);
    }
  }

  private void handleStatusRuntimeException(
      StatusRuntimeException e, StreamObserver<?> responseObserver) {
    if (e.getStatus().getCode() == Status.Code.INTERNAL) {
      log.error("unexpected", e);
    }
    responseObserver.onError(e);
  }

  @Override
  public void close() {}
}

1	/*
2	* Copyright (C) 2022 Temporal Technologies, Inc. All Rights Reserved.
3	*
4	* Copyright (C) 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
5	*
6	* Modifications copyright (C) 2017 Uber Technologies, Inc.
7	*
8	* Licensed under the Apache License, Version 2.0 (the "License");
9	* you may not use this material except in compliance with the License.
10	* You may obtain a copy of the License at
11	*
12	* http://www.apache.org/licenses/LICENSE-2.0
13	*
14	* Unless required by applicable law or agreed to in writing, software
15	* distributed under the License is distributed on an "AS IS" BASIS,
16	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
17	* See the License for the specific language governing permissions and
18	* limitations under the License.
19	*/
20
21	package io.temporal.internal.testservice;
22
23	import com.google.protobuf.ByteString;
24	import io.grpc.Status;
25	import io.grpc.StatusRuntimeException;
26	import io.grpc.stub.StreamObserver;
27	import io.temporal.api.enums.v1.IndexedValueType;
28	import io.temporal.api.nexus.v1.Endpoint;
29	import io.temporal.api.operatorservice.v1.*;
30	import java.io.Closeable;
31	import java.util.List;
32	import java.util.Map;
33	import org.slf4j.Logger;
34	import org.slf4j.LoggerFactory;
35
36	/**
37	* In memory implementation of the Operator Service. To be used for testing purposes only.
38	*
39	* <p>Do not use directly, instead use {@link io.temporal.testing.TestWorkflowEnvironment}.
40	*/
41	final class TestOperatorService extends OperatorServiceGrpc.OperatorServiceImplBase
42	implements Closeable {
43	private static final Logger log = LoggerFactory.getLogger(TestOperatorService.class);	1✔
44
45	private final TestVisibilityStore visibilityStore;
46	private final TestNexusEndpointStore nexusEndpointStore;
47
48	public TestOperatorService(
49	TestVisibilityStore visibilityStore, TestNexusEndpointStore nexusEndpointStore) {	1✔
50	this.visibilityStore = visibilityStore;	1✔
51	this.nexusEndpointStore = nexusEndpointStore;	1✔
52	}	1✔
53
54	@Override
55	public void addSearchAttributes(
56	AddSearchAttributesRequest request,
57	StreamObserver<AddSearchAttributesResponse> responseObserver) {
58	try {
59	Map<String, IndexedValueType> registeredSearchAttributes =	1✔
60	visibilityStore.getRegisteredSearchAttributes();	1✔
61	request.getSearchAttributesMap().keySet().stream()	1✔
62	.filter(registeredSearchAttributes::containsKey)	1✔
63	.findFirst()	1✔
64	.ifPresent(	1✔
65	sa -> {
66	throw Status.ALREADY_EXISTS	×
67	.withDescription("Search attribute " + sa + " already exists.")	×
68	.asRuntimeException();	×
69	});
70	request.getSearchAttributesMap().forEach(visibilityStore::addSearchAttribute);	1✔
71	responseObserver.onNext(AddSearchAttributesResponse.newBuilder().build());	1✔
72	responseObserver.onCompleted();	1✔
73	} catch (StatusRuntimeException e) {	×
74	handleStatusRuntimeException(e, responseObserver);	×
75	}	1✔
76	}	1✔
77
78	@Override
79	public void removeSearchAttributes(
80	RemoveSearchAttributesRequest request,
81	StreamObserver<RemoveSearchAttributesResponse> responseObserver) {
82	try {
83	Map<String, IndexedValueType> registeredSearchAttributes =	×
84	visibilityStore.getRegisteredSearchAttributes();	×
85	request.getSearchAttributesList().stream()	×
86	.filter(k -> !registeredSearchAttributes.containsKey(k))	×
87	.findFirst()	×
88	.ifPresent(	×
89	sa -> {
90	throw Status.NOT_FOUND	×
91	.withDescription("Search attribute " + sa + " doesn't exist.")	×
92	.asRuntimeException();	×
93	});
94	request.getSearchAttributesList().forEach(visibilityStore::removeSearchAttribute);	×
95	responseObserver.onNext(RemoveSearchAttributesResponse.newBuilder().build());	×
96	responseObserver.onCompleted();	×
97	} catch (StatusRuntimeException e) {	×
98	handleStatusRuntimeException(e, responseObserver);	×
99	}	×
100	}	×
101
102	@Override
103	public void getNexusEndpoint(
104	GetNexusEndpointRequest request, StreamObserver<GetNexusEndpointResponse> responseObserver) {
105	try {
106	Endpoint endpoint = nexusEndpointStore.getEndpoint(request.getId());	1✔
107	responseObserver.onNext(GetNexusEndpointResponse.newBuilder().setEndpoint(endpoint).build());	1✔
108	responseObserver.onCompleted();	1✔
109	} catch (StatusRuntimeException e) {	1✔
110	handleStatusRuntimeException(e, responseObserver);	1✔
111	}	1✔
112	}	1✔
113
114	@Override
115	public void createNexusEndpoint(
116	CreateNexusEndpointRequest request,
117	StreamObserver<CreateNexusEndpointResponse> responseObserver) {
118	try {
119	Endpoint created = nexusEndpointStore.createEndpoint(request.getSpec());	1✔
120	responseObserver.onNext(	1✔
121	CreateNexusEndpointResponse.newBuilder().setEndpoint(created).build());	1✔
122	responseObserver.onCompleted();	1✔
123	} catch (StatusRuntimeException e) {	1✔
124	handleStatusRuntimeException(e, responseObserver);	1✔
125	}	1✔
126	}	1✔
127
128	@Override
129	public void updateNexusEndpoint(
130	UpdateNexusEndpointRequest request,
131	StreamObserver<UpdateNexusEndpointResponse> responseObserver) {
132	try {
133	Endpoint updated =	1✔
134	nexusEndpointStore.updateEndpoint(	1✔
135	request.getId(), request.getVersion(), request.getSpec());	1✔
136	responseObserver.onNext(	1✔
137	UpdateNexusEndpointResponse.newBuilder().setEndpoint(updated).build());	1✔
138	responseObserver.onCompleted();	1✔
139	} catch (StatusRuntimeException e) {	1✔
140	handleStatusRuntimeException(e, responseObserver);	1✔
141	}	1✔
142	}	1✔
143
144	@Override
145	public void deleteNexusEndpoint(
146	DeleteNexusEndpointRequest request,
147	StreamObserver<DeleteNexusEndpointResponse> responseObserver) {
148	try {
149	nexusEndpointStore.deleteEndpoint(request.getId(), request.getVersion());	1✔
150	responseObserver.onNext(DeleteNexusEndpointResponse.newBuilder().build());	1✔
151	responseObserver.onCompleted();	1✔
152	} catch (StatusRuntimeException e) {	1✔
153	handleStatusRuntimeException(e, responseObserver);	1✔
154	}	1✔
155	}	1✔
156
157	@Override
158	public void listNexusEndpoints(
159	ListNexusEndpointsRequest request,
160	StreamObserver<ListNexusEndpointsResponse> responseObserver) {
161	try {
162	List<Endpoint> endpoints =	1✔
163	nexusEndpointStore.listEndpoints(	1✔
164	request.getPageSize(), request.getNextPageToken().toByteArray(), request.getName());	1✔
165	ByteString nextPageToken =
166	(!endpoints.isEmpty() && endpoints.size() == request.getPageSize())	1✔
167	? endpoints.get(endpoints.size() - 1).getIdBytes()	1✔
168	: ByteString.empty();	1✔
169	responseObserver.onNext(	1✔
170	ListNexusEndpointsResponse.newBuilder()	1✔
171	.addAllEndpoints(endpoints)	1✔
172	.setNextPageToken(nextPageToken)	1✔
173	.build());	1✔
174	responseObserver.onCompleted();	1✔
NEW 175	} catch (StatusRuntimeException e) {	×
NEW 176	handleStatusRuntimeException(e, responseObserver);	×
177	}	1✔
178	}	1✔
179
180	private void handleStatusRuntimeException(
181	StatusRuntimeException e, StreamObserver<?> responseObserver) {
182	if (e.getStatus().getCode() == Status.Code.INTERNAL) {	1✔
183	log.error("unexpected", e);	×
184	}
185	responseObserver.onError(e);	1✔
186	}	1✔
187
188	@Override
189	public void close() {}	1✔
190	}

temporalio / sdk-java / #294

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous