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
|
# -*- coding: utf-8 -*-
import os
import string
from hamcrest import assert_that, raises
from ydb.tests.library.harness.kikimr_cluster import kikimr_cluster_factory
from ydb.tests.library.harness.kikimr_config import KikimrConfigGenerator
from ydb.tests.library.harness.util import LogLevels
from ydb.tests.library.common.types import Erasure
import ydb
class Base(object):
erasure = None
@classmethod
def setup_class(cls):
cls.cluster = kikimr_cluster_factory(
KikimrConfigGenerator(
erasure=cls.erasure,
additional_log_configs={
'TENANT_POOL': LogLevels.DEBUG,
'KQP_PROXY': LogLevels.DEBUG
}
)
)
cls.cluster.start()
cls.database_name = "/Root"
cls.connection_params = ydb.DriverConfig("localhost:%d" % cls.cluster.nodes[1].port)
cls.connection_params.set_database(cls.database_name)
cls.driver = ydb.Driver(cls.connection_params)
cls.driver.wait()
@classmethod
def teardown_class(cls):
cls.driver.stop()
cls.cluster.stop()
class TestSchemeShardLimitsCase0(Base):
erasure = Erasure.BLOCK_4_2
def test_effective_acls_are_too_large(self):
# Arrange
directory_name = "/Root/test_effective_acls_are_too_large"
self.driver.scheme_client.make_directory(directory_name)
# Act
def callee():
for idx in string.ascii_lowercase:
current_directory = os.path.join(directory_name, idx)
self.driver.scheme_client.make_directory(current_directory)
subject = idx * (6 * 10 ** 6) + '@staff'
self.driver.scheme_client.modify_permissions(
current_directory,
ydb.ModifyPermissionsSettings()
.grant_permissions(
subject, (
'ydb.generic.read',
)
)
)
self.driver.scheme_client.describe_path(current_directory)
# Assert
assert_that(
callee,
raises(
ydb.GenericError,
"ACL is too long"
)
)
class TestSchemeShardLimitsCase1(Base):
erasure = Erasure.NONE
def test_too_large_acls(self):
# Arrange
directory_name = "/Root/test_too_large_acls"
self.driver.scheme_client.make_directory(directory_name)
# Act
def callee():
for chr_val in string.ascii_lowercase:
self.driver.scheme_client.modify_permissions(
directory_name,
ydb.ModifyPermissionsSettings()
.grant_permissions(
'gvit@staff', (
'ydb.generic.read',
)
)
.grant_permissions(
chr_val * 50 * 10 ** 6 + '@staff', (
'ydb.generic.read',
)
)
)
# Assert
assert_that(
callee,
raises(
ydb.GenericError,
"ACL is too long"
)
)
|
