|
| 1 | +import os |
| 2 | + |
1 | 3 | import pytest |
2 | 4 | import tt_helper |
3 | 5 |
|
@@ -109,3 +111,100 @@ def test_start_multi_inst(tt, tt_app, target): |
109 | 111 | ) |
110 | 112 | def test_start_cluster(tt, tt_app, target): |
111 | 113 | check_start(tt, tt_app, target) |
| 114 | + |
| 115 | + |
| 116 | +################################################################ |
| 117 | +# Permission denied |
| 118 | + |
| 119 | + |
| 120 | +@pytest.mark.skipif(skip_cluster_cond, reason=skip_cluster_reason) |
| 121 | +@pytest.mark.skipif( |
| 122 | + os.getuid() == 0, |
| 123 | + reason="Skipping the test, it shouldn't run as root", |
| 124 | +) |
| 125 | +@pytest.mark.tt_app(**tt_cluster_app) |
| 126 | +@pytest.mark.parametrize( |
| 127 | + "tt_running_targets", |
| 128 | + [ |
| 129 | + pytest.param([], id="running:none"), |
| 130 | + ], |
| 131 | +) |
| 132 | +@pytest.mark.parametrize( |
| 133 | + "denied_dir, expect_console_error", |
| 134 | + [ |
| 135 | + pytest.param(["var"], True, id="var_denied"), |
| 136 | + pytest.param(["var", "log"], True, id="var_log_denied"), |
| 137 | + pytest.param(["var", "lib"], False, id="var_lib_denied"), |
| 138 | + ], |
| 139 | +) |
| 140 | +def test_start_permission_denied(tt, tt_app, denied_dir, expect_console_error): |
| 141 | + target_dir = tt_app.path(*denied_dir) |
| 142 | + os.makedirs(target_dir, exist_ok=True) |
| 143 | + os.chmod(target_dir, 0o444) |
| 144 | + |
| 145 | + try: |
| 146 | + rc, out = tt.exec("start") |
| 147 | + |
| 148 | + if expect_console_error: |
| 149 | + assert rc != 0 |
| 150 | + assert "permission denied" in out.lower() |
| 151 | + else: |
| 152 | + assert rc == 0 |
| 153 | + |
| 154 | + logs = list(tt_helper.log_files(tt_app, tt_app.instances)) |
| 155 | + assert utils.wait_files(5, logs) |
| 156 | + |
| 157 | + error_found = False |
| 158 | + for log_file in logs: |
| 159 | + with open(log_file, "r") as f: |
| 160 | + if "permission denied" in f.read().lower(): |
| 161 | + error_found = True |
| 162 | + break |
| 163 | + |
| 164 | + assert error_found, "Permission denied error was not found in logs." |
| 165 | + finally: |
| 166 | + os.chmod(target_dir, 0o755) |
| 167 | + |
| 168 | + |
| 169 | +@pytest.mark.skipif(skip_cluster_cond, reason=skip_cluster_reason) |
| 170 | +@pytest.mark.skipif( |
| 171 | + os.getuid() == 0, |
| 172 | + reason="Skipping the test, it shouldn't run as root", |
| 173 | +) |
| 174 | +@pytest.mark.tt_app(**tt_cluster_app) |
| 175 | +@pytest.mark.parametrize( |
| 176 | + "tt_running_targets", |
| 177 | + [ |
| 178 | + pytest.param([], id="running:none"), |
| 179 | + ], |
| 180 | +) |
| 181 | +def test_start_permission_denied_permissions_denied_files(tt, tt_app): |
| 182 | + rc, _ = tt.exec("start") |
| 183 | + assert rc == 0 |
| 184 | + assert utils.wait_files(5, tt_helper.pid_files(tt_app, tt_app.instances)) |
| 185 | + |
| 186 | + rc, _ = tt.exec("stop", "-y") |
| 187 | + assert rc == 0 |
| 188 | + |
| 189 | + log_dir = tt_app.path("var", "log") |
| 190 | + lib_dir = tt_app.path("var", "lib") |
| 191 | + run_dir = tt_app.path("var", "run") |
| 192 | + |
| 193 | + dirs_to_lock = [] |
| 194 | + for data_dir in [log_dir, lib_dir, run_dir]: |
| 195 | + for root, dirs, files in os.walk(data_dir, topdown=False): |
| 196 | + dirs_to_lock.extend(os.path.join(root, f) for f in files) |
| 197 | + dirs_to_lock.extend(os.path.join(root, d) for d in dirs) |
| 198 | + dirs_to_lock.append(data_dir) |
| 199 | + |
| 200 | + for d in dirs_to_lock: |
| 201 | + os.chmod(d, 0o444) |
| 202 | + |
| 203 | + try: |
| 204 | + rc, out = tt.exec("start") |
| 205 | + assert rc != 0 |
| 206 | + assert "permission denied" in out.lower() |
| 207 | + finally: |
| 208 | + for d in dirs_to_lock: |
| 209 | + if os.path.exists(d): |
| 210 | + os.chmod(d, 0o755) |
0 commit comments