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
117
118
119
120
|
import unittest
import pytest
from influxdb_client import BucketRetentionRules
from influxdb_client.rest import ApiException
from tests.base_test import BaseTest, generate_bucket_name
class BucketsClientTest(BaseTest):
def setUp(self) -> None:
super(BucketsClientTest, self).setUp()
response = self.buckets_api.find_buckets()
for bucket in response.buckets:
if bucket.name.endswith("_IT"):
print("Delete bucket: ", bucket.name)
self.buckets_api.delete_bucket(bucket)
def test_create_delete_bucket(self):
my_org = self.find_my_org()
bucket_name = generate_bucket_name()
my_bucket = self.buckets_api.create_bucket(bucket_name=bucket_name, org=my_org)
self.assertEqual(my_bucket.name, bucket_name)
self.assertEqual(my_bucket.org_id, my_org.id)
print(my_bucket)
result = self.buckets_api.find_bucket_by_id(my_bucket.id)
print(result)
self.assertEqual(my_bucket, result)
self.delete_test_bucket(my_bucket)
with pytest.raises(ApiException) as e:
assert self.buckets_api.find_bucket_by_id(my_bucket.id)
assert "bucket not found" in e.value.body
def test_find_by_name(self):
my_org = self.find_my_org()
bucket_name = generate_bucket_name()
my_bucket = self.buckets_api.create_bucket(bucket_name=bucket_name, org=my_org)
bucket_by_name = self.buckets_api.find_bucket_by_name(bucket_name=my_bucket.name)
self.assertEqual(my_bucket, bucket_by_name)
none = self.buckets_api.find_bucket_by_name(bucket_name="non-existent-bucket")
self.assertIsNone(none)
self.buckets_api.delete_bucket(my_bucket)
def test_create_bucket_retention(self):
my_org = self.find_my_org()
bucket_name = generate_bucket_name()
retention = BucketRetentionRules(type="expire", every_seconds=3600)
desc = "bucket with retention"
my_bucket = self.buckets_api.create_bucket(bucket_name=bucket_name, org=my_org,
retention_rules=retention, description=desc)
self.assertEqual(my_bucket.description, desc)
print(my_bucket)
self.buckets_api.delete_bucket(my_bucket)
def test_create_bucket_retention_list(self):
my_org = self.find_my_org()
bucket_name = generate_bucket_name()
ret_list = []
retention = BucketRetentionRules(every_seconds=3600)
retention.type = "expire"
ret_list.append(retention)
my_bucket = self.buckets_api.create_bucket(bucket_name=bucket_name, org=my_org,
retention_rules=ret_list)
self.assertEqual(my_bucket.name, bucket_name)
self.delete_test_bucket(my_bucket)
def test_pagination(self):
my_org = self.find_my_org()
buckets = self.buckets_api.find_buckets().buckets
size = len(buckets)
# create 2 buckets
self.buckets_api.create_bucket(bucket_name=generate_bucket_name(), org=my_org)
self.buckets_api.create_bucket(bucket_name=generate_bucket_name(), org=my_org)
buckets = self.buckets_api.find_buckets().buckets
self.assertEqual(size + 2, len(buckets))
# offset 1
buckets = self.buckets_api.find_buckets(offset=1).buckets
self.assertEqual(size + 1, len(buckets))
# count 1
buckets = self.buckets_api.find_buckets(limit=1).buckets
self.assertEqual(1, len(buckets))
def test_update_bucket(self):
my_org = self.find_my_org()
bucket = self.buckets_api.create_bucket(bucket_name=generate_bucket_name(),
org=my_org,
description="my description")
self.assertEqual("my description", bucket.description)
bucket.description = "updated description"
self.buckets_api.update_bucket(bucket=bucket)
self.assertEqual("updated description", bucket.description)
if __name__ == '__main__':
unittest.main()
|
