Newer
Older
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
import random, logging, pytest, numpy
from dbscanserving.Config import GRPC_SERVICE_PORT, GRPC_MAX_WORKERS, GRPC_GRACE_PERIOD
from dbscanserving.client.DbscanServingClient import DbscanServingClient
from dbscanserving.service.DbscanService import DbscanService
from dbscanserving.proto.dbscanserving_pb2 import DetectionRequest, DetectionResponse, Sample
port = 10000 + GRPC_SERVICE_PORT # avoid privileged ports
LOGGER = logging.getLogger(__name__)
LOGGER.setLevel(logging.DEBUG)
@pytest.fixture(scope='session')
def dbscanserving_service():
_service = DbscanService(
port=port, max_workers=GRPC_MAX_WORKERS, grace_period=GRPC_GRACE_PERIOD)
_service.start()
yield _service
_service.stop()
@pytest.fixture(scope='session')
def dbscanserving_client():
_client = DbscanServingClient(address='127.0.0.1', port=port)
yield _client
_client.close()
def test_detection_correct(dbscanserving_service, dbscanserving_client: DbscanServingClient):
request: DetectionRequest = DetectionRequest()
request.num_samples = 310
request.num_features = 100
request.eps = 100.5
request.min_samples = 50
for _ in range(200):
grpc_sample = Sample()
for __ in range(100):
grpc_sample.features.append(random.uniform(0., 10.))
request.samples.append(grpc_sample)
for _ in range(100):
grpc_sample = Sample()
for __ in range(100):
grpc_sample.features.append(random.uniform(50., 60.))
request.samples.append(grpc_sample)
for _ in range(10):
grpc_sample = Sample()
for __ in range(100):
grpc_sample.features.append(random.uniform(5000., 6000.))
request.samples.append(grpc_sample)
response: DetectionResponse = dbscanserving_client.Detect(request)
assert len(response.cluster_indices) == 310
def test_detection_incorrect(dbscanserving_service, dbscanserving_client: DbscanServingClient):
request: DetectionRequest = DetectionRequest()
request.num_samples = 210
request.num_features = 100
request.eps = 100.5
request.min_samples = 50
for _ in range(200):
grpc_sample = Sample()
for __ in range(100):
grpc_sample.features.append(random.uniform(0., 10.))
request.samples.append(grpc_sample)
for _ in range(100):
grpc_sample = Sample()
for __ in range(100):
grpc_sample.features.append(random.uniform(50., 60.))
request.samples.append(grpc_sample)
for _ in range(10):
grpc_sample = Sample()
for __ in range(100):
grpc_sample.features.append(random.uniform(5000., 6000.))
request.samples.append(grpc_sample)
with pytest.raises(Exception):
response: DetectionResponse = dbscanserving_client.Detect(request)
def test_detection_clusters(dbscanserving_service, dbscanserving_client: DbscanServingClient):
request: DetectionRequest = DetectionRequest()
request.num_samples = 310
request.num_features = 100
request.eps = 100.5
request.min_samples = 50
for _ in range(200):
grpc_sample = Sample()
for __ in range(100):
grpc_sample.features.append(random.uniform(0., 10.))
request.samples.append(grpc_sample)
for _ in range(100):
grpc_sample = Sample()
for __ in range(100):
grpc_sample.features.append(random.uniform(50., 60.))
request.samples.append(grpc_sample)
for _ in range(10):
grpc_sample = Sample()
for __ in range(100):
grpc_sample.features.append(random.uniform(5000., 6000.))
request.samples.append(grpc_sample)
response: DetectionResponse = dbscanserving_client.Detect(request)
for v in response.cluster_indices[:200]:
assert v == 0
for v in response.cluster_indices[200:300]:
assert v == 1
for v in response.cluster_indices[300:]:
assert v == -1