Coverage for src/meshadmin/server/networks/tests/test_views.py: 100%

1from datetime import timedelta 

2 

3import pytest 

4from django.urls import reverse 

5from django.utils import timezone 

6 

7from meshadmin.common.utils import create_keys 

8from meshadmin.server.networks.models import ( 

9 CA, 

10 ConfigRollout, 

11 Group, 

12 GroupConfig, 

13 Host, 

14 HostConfig, 

15 Network, 

16 Rule, 

17 Template, 

18) 

19from meshadmin.server.networks.services import create_group, create_template 

20 

21 

22class TestRolloutViews: 

23 def test_rollout_creation_with_hosts(self, auth_client, test_network): 

24 client, user = auth_client() 

25 test_net = test_network(name="testnet", cidr="100.100.64.0/24", user=user) 

26 hosts = [] 

27 for i in range(3): 

28 host = Host.objects.create( 

29 network=test_net, 

30 name=f"test-host-{i}", 

31 assigned_ip=f"100.100.64.{i + 1}", 

32 ) 

33 hosts.append(host) 

34 

35 data = { 

36 "name": "new-rollout", 

37 "notes": "Test notes", 

38 "hosts": [host.id for host in hosts[:2]], 

39 } 

40 

41 response = client.post( 

42 reverse( 

43 "networks:network-rollout-create", kwargs={"network_id": test_net.id} 

44 ), 

45 data, 

46 ) 

47 

48 assert response.status_code == 302 

49 rollout = ConfigRollout.objects.get(name="new-rollout") 

50 assert rollout.network == test_net 

51 assert rollout.notes == "Test notes" 

52 assert rollout.status == "PENDING" 

53 assert list(rollout.target_hosts.all()) == hosts[:2] 

54 for host in hosts[:2]: 

55 host.refresh_from_db() 

56 assert host.config_freeze is True 

57 

58 def test_rollout_unfreeze(self, auth_client, test_network): 

59 client, user = auth_client() 

60 test_net = test_network(name="testnet", cidr="100.100.64.0/24", user=user) 

61 hosts = [] 

62 for i in range(2): 

63 host = Host.objects.create( 

64 network=test_net, 

65 name=f"test-host-{i}", 

66 assigned_ip=f"100.100.64.{i + 1}", 

67 config_freeze=True, 

68 ) 

69 hosts.append(host) 

70 

71 rollout = ConfigRollout.objects.create( 

72 name="test-rollout", 

73 network=test_net, 

74 status="PENDING", 

75 ) 

76 rollout.target_hosts.add(*hosts) 

77 

78 # Test single host unfreeze 

79 response = client.post( 

80 reverse("networks:rollout-unfreeze", kwargs={"pk": rollout.pk}), 

81 {"host_id": hosts[0].id}, 

82 ) 

83 

84 assert response.status_code == 302 

85 rollout.refresh_from_db() 

86 assert rollout.status == "PENDING" 

87 assert list(rollout.completed_hosts.all()) == [hosts[0]] 

88 hosts[0].refresh_from_db() 

89 assert hosts[0].config_freeze is False 

90 hosts[1].refresh_from_db() 

91 assert hosts[1].config_freeze is True 

92 

93 # Test complete rollout unfreeze 

94 response = client.post( 

95 reverse("networks:rollout-unfreeze", kwargs={"pk": rollout.pk}), 

96 ) 

97 assert response.status_code == 302 

98 rollout.refresh_from_db() 

99 assert rollout.status == "COMPLETED" 

100 assert set(rollout.completed_hosts.all()) == set(hosts) 

101 for host in hosts: 

102 host.refresh_from_db() 

103 assert host.config_freeze is False 

104 

105 def test_rollout_update(self, auth_client, test_network): 

106 client, user = auth_client() 

107 test_net = test_network(name="testnet", cidr="100.100.64.0/24", user=user) 

108 hosts = [] 

109 for i in range(3): 

110 host = Host.objects.create( 

111 network=test_net, 

112 name=f"test-host-{i}", 

113 assigned_ip=f"100.100.64.{i + 1}", 

114 ) 

115 hosts.append(host) 

116 

117 rollout = ConfigRollout.objects.create( 

118 name="test-rollout", 

119 network=test_net, 

120 status="PENDING", 

121 ) 

122 rollout.target_hosts.add(*hosts[:2]) 

123 

124 data = { 

125 "name": "updated-rollout", 

126 "notes": "Updated notes", 

127 "hosts": [hosts[0].id, hosts[2].id], 

128 } 

129 

130 response = client.post( 

131 reverse("networks:rollout-edit", kwargs={"pk": rollout.pk}), 

132 data, 

133 ) 

134 

135 assert response.status_code == 302 

136 rollout.refresh_from_db() 

137 assert rollout.name == "updated-rollout" 

138 assert rollout.notes == "Updated notes" 

139 assert list(rollout.target_hosts.all()) == [hosts[0], hosts[2]] 

140 hosts[0].refresh_from_db() 

141 hosts[1].refresh_from_db() 

142 hosts[2].refresh_from_db() 

143 assert hosts[0].config_freeze is True 

144 assert hosts[1].config_freeze is False 

145 assert hosts[2].config_freeze is True 

146 

147 def test_rollout_delete(self, auth_client, test_network): 

148 client, user = auth_client() 

149 test_net = test_network(name="testnet", cidr="100.100.64.0/24", user=user) 

150 hosts = [] 

151 for i in range(2): 

152 host = Host.objects.create( 

153 network=test_net, 

154 name=f"test-host-{i}", 

155 assigned_ip=f"100.100.64.{i + 1}", 

156 config_freeze=True, 

157 ) 

158 hosts.append(host) 

159 

160 rollout = ConfigRollout.objects.create( 

161 name="test-rollout", 

162 network=test_net, 

163 status="PENDING", 

164 ) 

165 rollout.target_hosts.add(*hosts) 

166 

167 response = client.post( 

168 reverse("networks:rollout-delete", kwargs={"pk": rollout.pk}), 

169 ) 

170 

171 assert response.status_code == 302 

172 assert not ConfigRollout.objects.filter(pk=rollout.pk).exists() 

173 for host in hosts: 

174 host.refresh_from_db() 

175 assert host.config_freeze is False 

176 

177 

178class TestHostViews: 

179 def test_host_refresh_config_with_rollout(self, auth_client, test_network): 

180 client, user = auth_client() 

181 test_net = test_network(name="testnet", cidr="100.100.64.0/24", user=user) 

182 _, public_key = create_keys() 

183 host = Host.objects.create( 

184 network=test_net, 

185 name="test-host", 

186 assigned_ip="100.100.64.1", 

187 config_freeze=True, 

188 public_key=public_key, 

189 interface="nebula1", 

190 ) 

191 rollout = ConfigRollout.objects.create( 

192 name="test-rollout", 

193 network=test_net, 

194 status="PENDING", 

195 ) 

196 rollout.target_hosts.add(host) 

197 response = client.post( 

198 reverse( 

199 "networks:host-refresh-config", 

200 kwargs={"pk": host.pk, "rollout_id": rollout.pk}, 

201 ), 

202 ) 

203 assert response.status_code == 302 

204 assert response.url == reverse( 

205 "networks:rollout-detail", kwargs={"pk": rollout.pk} 

206 ) 

207 host.refresh_from_db() 

208 assert host.hostconfig_set.count() > 0 

209 latest_config = host.hostconfig_set.latest("created_at") 

210 assert "pki" in latest_config.config 

211 assert "lighthouse" in latest_config.config 

212 

213 def test_config_diff_view(self, auth_client, test_network): 

214 client, user = auth_client() 

215 test_net = test_network(name="testnet", cidr="100.100.64.0/24", user=user) 

216 host = Host.objects.create(name="test_host", network=test_net) 

217 

218 # Create two configs with different content 

219 config1 = HostConfig.objects.create( 

220 host=host, config="test: config1\nline: 1", sha256="abc123" 

221 ) 

222 config2 = HostConfig.objects.create( 

223 host=host, config="test: config2\nline: 2", sha256="def456" 

224 ) 

225 

226 # Test successful diff 

227 response = client.get( 

228 reverse( 

229 "networks:config-diff", 

230 kwargs={"base_id": config1.id, "compare_id": config2.id}, 

231 ) 

232 ) 

233 assert response.status_code == 200 

234 assert b"-test: config1" in response.content 

235 assert b"+test: config2" in response.content 

236 assert b"-line: 1" in response.content 

237 assert b"+line: 2" in response.content 

238 

239 # Test no changes between identical configs 

240 config3 = HostConfig.objects.create( 

241 host=host, config="test: config1\nline: 1", sha256="abc123" 

242 ) 

243 response = client.get( 

244 reverse( 

245 "networks:config-diff", 

246 kwargs={"base_id": config1.id, "compare_id": config3.id}, 

247 ) 

248 ) 

249 assert response.status_code == 200 

250 assert b"No differences found" in response.content 

251 

252 # Test non-existent config 

253 response = client.get( 

254 reverse( 

255 "networks:config-diff", 

256 kwargs={"base_id": 99999, "compare_id": config2.id}, 

257 ) 

258 ) 

259 assert response.status_code == 200 

260 assert b"Error" in response.content 

261 

262 def test_make_signing_ca(self, auth_client, test_network): 

263 client, user = auth_client() 

264 network = test_network(name="testnet", cidr="100.100.64.0/24", user=user) 

265 new_ca = CA.objects.create( 

266 network=network, 

267 name="new_signing_ca", 

268 key="test_key", 

269 cert="test_cert", 

270 ) 

271 original_signing_ca = network.signingca.ca 

272 response = client.post( 

273 reverse("networks:ca-make-signing", kwargs={"pk": new_ca.pk}), 

274 ) 

275 assert response.status_code == 200 

276 network.refresh_from_db() 

277 assert network.signingca.ca == new_ca 

278 assert network.signingca.ca != original_signing_ca 

279 

280 

281class TestCRUDWithParentNetwork: 

282 @pytest.mark.parametrize( 

283 "url_name,data,expected_model,expected_fields", 

284 [ 

285 ( 

286 "networks:network-ca-create", 

287 {"name": "test_ca"}, 

288 CA, 

289 {"name": "test_ca", "cert__isnull": False, "key__isnull": False}, 

290 ), 

291 ( 

292 "networks:network-group-create", 

293 {"name": "test_group"}, 

294 Group, 

295 {"name": "test_group"}, 

296 ), 

297 ], 

298 ) 

299 def test_entity_creation_with_parent_network( 

300 self, 

301 auth_client, 

302 test_network, 

303 url_name, 

304 data, 

305 expected_model, 

306 expected_fields, 

307 ): 

308 client, user = auth_client() 

309 test_net = test_network(name="testnet", cidr="100.100.64.0/24", user=user) 

310 data["network"] = test_net.id 

311 

312 response = client.post( 

313 reverse(url_name, kwargs={"network_id": test_net.id}), 

314 data, 

315 ) 

316 assert response.status_code == 302 

317 filters = {**expected_fields, "network": test_net} 

318 assert expected_model.objects.filter(**filters).exists() 

319 

320 @pytest.mark.parametrize( 

321 "url_name,model,update_data", 

322 [ 

323 ("networks:ca-edit", CA, {"name": "updated_name"}), 

324 ("networks:group-edit", Group, {"name": "updated_name"}), 

325 ], 

326 ) 

327 def test_entity_update_with_parent_network( 

328 self, auth_client, test_network, url_name, model, update_data 

329 ): 

330 client, user = auth_client() 

331 test_net = test_network(name="testnet", cidr="100.100.64.0/24", user=user) 

332 

333 obj = model.objects.create( 

334 network=test_net, name=f"test_{model._meta.model_name}" 

335 ) 

336 update_data["network"] = test_net.id 

337 

338 response = client.post( 

339 reverse(url_name, kwargs={"pk": obj.id}), 

340 update_data, 

341 ) 

342 assert response.status_code == 302 

343 obj.refresh_from_db() 

344 assert obj.name == update_data["name"] 

345 

346 @pytest.mark.parametrize( 

347 "url_name,model", 

348 [ 

349 ("networks:ca-delete", CA), 

350 ("networks:group-delete", Group), 

351 ], 

352 ) 

353 def test_entity_deletion_with_parent_network( 

354 self, auth_client, test_network, url_name, model 

355 ): 

356 client, user = auth_client() 

357 test_net = test_network(name="testnet", cidr="100.100.64.0/24", user=user) 

358 obj = model.objects.create( 

359 network=test_net, 

360 name=f"test_{model._meta.model_name}", 

361 ) 

362 response = client.post( 

363 reverse(url_name, kwargs={"pk": obj.id}), 

364 ) 

365 assert response.status_code == 302 

366 assert not model.objects.filter(id=obj.id).exists() 

367 

368 

369class TestRuleViews: 

370 def test_add_rule_to_group_success(self, auth_client, test_network): 

371 client, user = auth_client() 

372 test_net = test_network(name="testnet", cidr="100.100.64.0/24", user=user) 

373 

374 # First create a security group 

375 group_data = {"name": "test_group", "description": "Test security group"} 

376 response = client.post( 

377 reverse( 

378 "networks:network-group-create", kwargs={"network_id": test_net.id} 

379 ), 

380 group_data, 

381 ) 

382 assert response.status_code == 302 

383 security_group = Group.objects.get(name="test_group") 

384 

385 # create target groups 

386 target_group1 = create_group(test_net.pk, "target_group1") 

387 target_group2 = create_group(test_net.pk, "target_group2") 

388 target_group3 = create_group(test_net.pk, "target_group3") 

389 

390 # add a rule to the group 

391 rule_data = { 

392 "security_group": security_group.id, 

393 "direction": "I", 

394 "proto": "tcp", 

395 "port": "80", 

396 "cidr": "0.0.0.0/0", 

397 "group": target_group1.id, 

398 "groups": [target_group2.id, target_group3.id], 

399 "local_cidr": "192.168.1.0/24", 

400 } 

401 

402 response = client.post( 

403 reverse("networks:group-add-rule"), 

404 rule_data, 

405 ) 

406 

407 assert response.status_code == 200 

408 assert Rule.objects.filter(security_group=security_group).exists() 

409 rule = Rule.objects.get(security_group=security_group) 

410 assert rule.direction == "I" 

411 assert rule.proto == "tcp" 

412 assert rule.port == "80" 

413 assert rule.cidr == "0.0.0.0/0" 

414 assert rule.group == target_group1 

415 assert set(rule.groups.all()) == {target_group2, target_group3} 

416 assert rule.local_cidr == "192.168.1.0/24" 

417 

418 def test_add_rule_validation_no_target(self, auth_client, test_network): 

419 client, user = auth_client() 

420 test_net = test_network(name="testnet", cidr="100.100.64.0/24", user=user) 

421 group = create_group(test_net.pk, "test_group") 

422 rule_data = { 

423 "security_group": group.id, 

424 "direction": "I", 

425 "proto": "tcp", 

426 "port": "80", 

427 } 

428 response = client.post( 

429 reverse("networks:group-add-rule"), 

430 rule_data, 

431 ) 

432 assert response.status_code == 200 

433 assert "At least one of group, groups, or CIDR" in response.content.decode() 

434 assert not Rule.objects.filter(security_group=group).exists() 

435 

436 def test_add_rule_validation_invalid_port(self, auth_client, test_network): 

437 client, user = auth_client() 

438 test_net = test_network(name="testnet", cidr="100.100.64.0/24", user=user) 

439 group = create_group(test_net.pk, "test_group") 

440 rule_data = { 

441 "security_group": group.id, 

442 "direction": "I", 

443 "proto": "tcp", 

444 "port": "80-invalid", 

445 "cidr": "0.0.0.0/0", 

446 } 

447 

448 response = client.post( 

449 reverse("networks:group-add-rule"), 

450 rule_data, 

451 ) 

452 

453 assert response.status_code == 200 

454 assert "Port range must be two" in response.content.decode() 

455 assert not Rule.objects.filter(security_group=group).exists() 

456 

457 # Port out of range 

458 rule_data["port"] = "70000" 

459 response = client.post( 

460 reverse("networks:group-add-rule"), 

461 rule_data, 

462 ) 

463 assert response.status_code == 200 

464 assert "Port must be" in response.content.decode() 

465 assert not Rule.objects.filter(security_group=group).exists() 

466 

467 def test_add_rule_validation_invalid_cidr(self, auth_client, test_network): 

468 client, user = auth_client() 

469 test_net = test_network(name="testnet", cidr="100.100.64.0/24", user=user) 

470 group = create_group(test_net.pk, "test_group") 

471 rule_data = { 

472 "security_group": group.id, 

473 "direction": "I", 

474 "proto": "tcp", 

475 "port": "80", 

476 "cidr": "invalid-cidr", 

477 } 

478 response = client.post( 

479 reverse("networks:group-add-rule"), 

480 rule_data, 

481 ) 

482 assert response.status_code == 200 

483 assert "Invalid CIDR format" in response.content.decode() 

484 assert not Rule.objects.filter(security_group=group).exists() 

485 

486 # Invalid local CIDR format 

487 rule_data = { 

488 "security_group": group.id, 

489 "direction": "I", 

490 "proto": "tcp", 

491 "port": "80", 

492 "cidr": "0.0.0.0/0", 

493 "local_cidr": "invalid-local-cidr", 

494 } 

495 response = client.post( 

496 reverse("networks:group-add-rule"), 

497 rule_data, 

498 ) 

499 assert response.status_code == 200 

500 assert "Invalid CIDR format" in response.content.decode() 

501 assert not Rule.objects.filter(security_group=group).exists() 

502 

503 

504class TestNetworkViews: 

505 def test_network_list(self, auth_client, test_network): 

506 client, user = auth_client() 

507 test_net = test_network(name="testnet", cidr="100.100.64.0/24", user=user) 

508 response = client.get(reverse("networks:network-list")) 

509 assert response.status_code == 200 

510 assert test_net.name.encode() in response.content 

511 assert test_net.cidr.encode() in response.content 

512 

513 def test_network_detail(self, auth_client, test_network): 

514 client, user = auth_client() 

515 test_net = test_network(name="testnet", cidr="100.100.64.0/24", user=user) 

516 ca = CA.objects.create(network=test_net, name="test_ca") 

517 group = create_group(test_net.pk, "test_group") 

518 

519 response = client.get( 

520 reverse("networks:network-detail", kwargs={"pk": test_net.id}) 

521 ) 

522 

523 assert response.status_code == 200 

524 assert test_net.name.encode() in response.content 

525 assert test_net.cidr.encode() in response.content 

526 assert ca.name.encode() in response.content 

527 assert group.name.encode() in response.content 

528 

529 @pytest.mark.parametrize( 

530 "test_data,expected_status", 

531 [ 

532 ({"name": "test_net", "cidr": "192.168.1.0/24", "update_interval": 5}, 302), 

533 ({"name": "test_net", "cidr": "10.0.0.0/16", "update_interval": 5}, 302), 

534 ({"name": "test_net", "cidr": "100.64.0.0/24", "update_interval": 5}, 302), 

535 ( 

536 {"name": "test_net", "cidr": "199.100.69.0/24", "update_interval": 5}, 

537 200, 

538 ), 

539 ], 

540 ) 

541 def test_network_cidr_validation(self, auth_client, test_data, expected_status): 

542 client, _ = auth_client() 

543 response = client.post(reverse("networks:network-create"), test_data) 

544 

545 assert response.status_code == expected_status 

546 

547 if expected_status == 302: 

548 assert Network.objects.filter(name=test_data["name"]).exists() 

549 else: 

550 assert not Network.objects.filter(name=test_data["name"]).exists() 

551 

552 

553class TestTemplateViews: 

554 def test_template_creation_with_security_group(self, auth_client, test_network): 

555 client, user = auth_client() 

556 test_net = test_network(name="testnet", cidr="100.100.64.0/24", user=user) 

557 group1 = create_group(test_net.pk, "group1") 

558 group2 = create_group(test_net.pk, "group2") 

559 security_group = create_group( 

560 test_net.pk, 

561 "security_group", 

562 "Security group for testing", 

563 ) 

564 Rule.objects.create( 

565 security_group=security_group, 

566 group=group1, 

567 direction="I", 

568 proto="tcp", 

569 port="80", 

570 ) 

571 data = { 

572 "name": "test_template", 

573 "network": test_net.id, 

574 "is_lighthouse": False, 

575 "is_relay": False, 

576 "use_relay": True, 

577 "groups": [group1.id, group2.id, security_group.id], 

578 } 

579 response = client.post( 

580 reverse( 

581 "networks:network-template-create", kwargs={"network_id": test_net.id} 

582 ), 

583 data, 

584 ) 

585 assert response.status_code == 302 

586 template = Template.objects.get(name="test_template") 

587 assert list(template.groups.all()) == [group1, group2, security_group] 

588 

589 def test_template_deletion(self, auth_client, test_network): 

590 client, user = auth_client() 

591 test_net = test_network(name="testnet", cidr="100.100.64.0/24", user=user) 

592 security_group = create_group( 

593 test_net.pk, "test_security_group", "Test security group" 

594 ) 

595 template = create_template( 

596 "test_template", 

597 test_net.name, 

598 groups=[security_group.name], 

599 ) 

600 

601 response = client.post( 

602 reverse("networks:template-delete", kwargs={"pk": template.id}), 

603 ) 

604 

605 assert response.status_code == 302 

606 assert not Template.objects.filter(id=template.id).exists() 

607 assert Group.objects.filter(id=security_group.id).exists() 

608 

609 def test_template_creation_with_all_settings(self, auth_client, test_network): 

610 client, user = auth_client() 

611 test_net = test_network(name="testnet", cidr="100.100.64.0/24", user=user) 

612 data = { 

613 "name": "test_template_with_settings", 

614 "network": test_net.id, 

615 "is_lighthouse": False, 

616 "is_relay": False, 

617 "use_relay": True, 

618 "reusable": False, 

619 "usage_limit": 5, 

620 "expiry_days": 30, 

621 "ephemeral_peers": True, 

622 } 

623 

624 response = client.post( 

625 reverse( 

626 "networks:network-template-create", kwargs={"network_id": test_net.id} 

627 ), 

628 data, 

629 ) 

630 

631 assert response.status_code == 302 

632 template = Template.objects.get(name="test_template_with_settings") 

633 assert template.reusable is False 

634 assert template.usage_limit == 5 

635 assert template.ephemeral_peers is True 

636 assert template.expires_at is not None 

637 

638 # The expiry_days field should be converted to an expires_at datetime 

639 # Check that it's approximately 30 days in the future (within 1 day tolerance) 

640 expected_expiry = timezone.now() + timedelta(days=30) 

641 assert ( 

642 abs((template.expires_at - expected_expiry).total_seconds()) < 86400 

643 ) # 1 day in seconds 

644 

645 def test_template_update_with_enrollment_settings(self, auth_client, test_network): 

646 client, user = auth_client() 

647 test_net = test_network(name="testnet", cidr="100.100.64.0/24", user=user) 

648 template = create_template( 

649 "template_to_update", 

650 test_net.name, 

651 reusable=True, 

652 usage_limit=None, 

653 ephemeral_peers=False, 

654 ) 

655 data = { 

656 "name": "updated_template", 

657 "network": test_net.id, 

658 "is_lighthouse": False, 

659 "is_relay": False, 

660 "use_relay": True, 

661 "reusable": False, 

662 "usage_limit": 10, 

663 "expiry_days": 15, 

664 "ephemeral_peers": True, 

665 } 

666 response = client.post( 

667 reverse("networks:template-edit", kwargs={"pk": template.id}), 

668 data, 

669 ) 

670 assert response.status_code == 302 

671 template.refresh_from_db() 

672 assert template.name == "updated_template" 

673 assert template.reusable is False 

674 assert template.usage_limit == 10 

675 assert template.ephemeral_peers is True 

676 assert template.expires_at is not None 

677 

678 # Check expiry date is approximately 15 days in the future 

679 expected_expiry = timezone.now() + timedelta(days=15) 

680 assert abs((template.expires_at - expected_expiry).total_seconds()) < 86400 

681 

682 def test_template_update_remove_expiry(self, auth_client, test_network): 

683 client, user = auth_client() 

684 test_net = test_network(name="testnet", cidr="100.100.64.0/24", user=user) 

685 template = Template.objects.create( 

686 name="template_with_expiry", 

687 network=test_net, 

688 expires_at=timezone.now() + timedelta(days=30), 

689 ) 

690 data = { 

691 "name": "template_without_expiry", 

692 "network": test_net.id, 

693 "is_lighthouse": False, 

694 "is_relay": False, 

695 "use_relay": True, 

696 "reusable": True, 

697 "usage_limit": "", # Empty string for no limit 

698 "expiry_days": "", # Empty string for no expiration 

699 "ephemeral_peers": False, 

700 } 

701 response = client.post( 

702 reverse("networks:template-edit", kwargs={"pk": template.id}), 

703 data, 

704 ) 

705 assert response.status_code == 302 

706 template.refresh_from_db() 

707 assert template.name == "template_without_expiry" 

708 assert template.expires_at is None 

709 

710 

711class TestNetworkMembershipViews: 

712 def test_add_member(self, auth_client, test_network, create_user): 

713 client, admin_user = auth_client() 

714 network = test_network(name="testnet", cidr="100.100.64.0/24", user=admin_user) 

715 new_member = create_user(email="member@example.com", username="member") 

716 url = reverse("networks:network-member-add", kwargs={"network_id": network.id}) 

717 data = { 

718 "email": new_member.email, 

719 "role": "MEMBER", 

720 } 

721 response = client.post(url, data) 

722 assert response.status_code == 302 

723 membership = network.memberships.get(user=new_member) 

724 assert membership.role == "MEMBER" 

725 

726 def test_add_duplicate_member(self, auth_client, test_network, create_user): 

727 client, admin_user = auth_client() 

728 network = test_network(name="testnet", cidr="100.100.64.0/24", user=admin_user) 

729 existing_member = create_user(email="member@example.com", username="member") 

730 network.memberships.create(user=existing_member, role="MEMBER") 

731 url = reverse("networks:network-member-add", kwargs={"network_id": network.id}) 

732 data = { 

733 "email": existing_member.email, 

734 "role": "MEMBER", 

735 } 

736 response = client.post(url, data) 

737 assert response.status_code == 200 

738 assert ( 

739 "This user is already a member of the network" in response.content.decode() 

740 ) 

741 

742 def test_edit_member_role(self, auth_client, test_network, create_user): 

743 client, admin_user = auth_client() 

744 network = test_network(name="testnet", cidr="100.100.64.0/24", user=admin_user) 

745 member = create_user(email="member@example.com", username="member")