11# Use random integer as the start point here to avoid
22# id conflicts when multiple testings are running.
3+ import ipaddress
34import json
45import logging
56import os
1314from typing import Callable , Optional
1415
1516import pytest
17+ import requests
1618
1719from linodecli import ENV_TOKEN_NAME
1820from tests .integration .helpers import (
3335NODEBALANCER_BASE_CMD = ["linode-cli" , "nodebalancers" ]
3436
3537
38+ @pytest .fixture (autouse = True , scope = "session" )
39+ def linode_cloud_firewall ():
40+ def is_valid_ipv4 (address ):
41+ try :
42+ ipaddress .IPv4Address (address )
43+ return True
44+ except ipaddress .AddressValueError :
45+ return False
46+
47+ def is_valid_ipv6 (address ):
48+ try :
49+ ipaddress .IPv6Address (address )
50+ return True
51+ except ipaddress .AddressValueError :
52+ return False
53+
54+ def get_public_ip (ip_version = "ipv4" ):
55+ url = (
56+ f"https://api64.ipify.org?format=json"
57+ if ip_version == "ipv6"
58+ else f"https://api.ipify.org?format=json"
59+ )
60+ response = requests .get (url )
61+ return str (response .json ()["ip" ])
62+
63+ def create_inbound_rule (ipv4_address , ipv6_address ):
64+ rule = [
65+ {
66+ "protocol" : "TCP" ,
67+ "ports" : "22" ,
68+ "addresses" : {},
69+ "action" : "ACCEPT" ,
70+ }
71+ ]
72+ if is_valid_ipv4 (ipv4_address ):
73+ rule [0 ]["addresses" ]["ipv4" ] = [f"{ ipv4_address } /32" ]
74+
75+ if is_valid_ipv6 (ipv6_address ):
76+ rule [0 ]["addresses" ]["ipv6" ] = [f"{ ipv6_address } /128" ]
77+
78+ return json .dumps (rule , indent = 4 )
79+
80+ # Fetch the public IP addresses
81+ ipv4_address = get_public_ip ("ipv4" )
82+ ipv6_address = get_public_ip ("ipv6" )
83+
84+ inbound_rule = create_inbound_rule (ipv4_address , ipv6_address )
85+
86+ label = "cloud_firewall_" + str (int (time .time ()))
87+
88+ # Base command list
89+ command = [
90+ "linode-cli" ,
91+ "firewalls" ,
92+ "create" ,
93+ "--label" ,
94+ label ,
95+ "--rules.outbound_policy" ,
96+ "ACCEPT" ,
97+ "--rules.inbound_policy" ,
98+ "DROP" ,
99+ "--text" ,
100+ "--no-headers" ,
101+ "--format" ,
102+ "id" ,
103+ ]
104+
105+ if is_valid_ipv4 (ipv4_address ) or is_valid_ipv6 (ipv6_address ):
106+ command .extend (["--rules.inbound" , inbound_rule ])
107+
108+ firewall_id = exec_test_command (command ).stdout .decode ().rstrip ()
109+
110+ yield firewall_id
111+
112+ delete_target_id (target = "firewalls" , id = firewall_id )
113+
114+
36115@pytest .fixture (scope = "session" )
37116def _id_generators ():
38117 return defaultdict (lambda : count (randint (0 , 1000000 )))
@@ -185,7 +264,7 @@ def slave_domain():
185264
186265# Test helpers specific to Linodes test suite
187266@pytest .fixture
188- def linode_with_label ():
267+ def linode_with_label (linode_cloud_firewall ):
189268 timestamp = str (time .time_ns ())
190269 label = "cli" + timestamp
191270 result = (
@@ -203,6 +282,8 @@ def linode_with_label():
203282 label ,
204283 "--root_pass" ,
205284 DEFAULT_RANDOM_PASS ,
285+ "--firewall_id" ,
286+ linode_cloud_firewall ,
206287 "--text" ,
207288 "--delimiter" ,
208289 "," ,
@@ -223,7 +304,7 @@ def linode_with_label():
223304
224305
225306@pytest .fixture
226- def linode_min_req ():
307+ def linode_min_req (linode_cloud_firewall ):
227308 result = (
228309 exec_test_command (
229310 LINODE_BASE_CMD
@@ -235,6 +316,8 @@ def linode_min_req():
235316 "us-ord" ,
236317 "--root_pass" ,
237318 DEFAULT_RANDOM_PASS ,
319+ "--firewall_id" ,
320+ linode_cloud_firewall ,
238321 "--no-defaults" ,
239322 "--text" ,
240323 "--delimiter" ,
@@ -256,7 +339,7 @@ def linode_min_req():
256339
257340
258341@pytest .fixture
259- def linode_wo_image ():
342+ def linode_wo_image (linode_cloud_firewall ):
260343 label = "cli" + str (int (time .time ()) + randint (10 , 1000 ))
261344 linode_id = (
262345 exec_test_command (
@@ -272,6 +355,8 @@ def linode_wo_image():
272355 DEFAULT_REGION ,
273356 "--root_pass" ,
274357 DEFAULT_RANDOM_PASS ,
358+ "--firewall_id" ,
359+ linode_cloud_firewall ,
275360 "--format" ,
276361 "id" ,
277362 "--no-headers" ,
@@ -288,7 +373,7 @@ def linode_wo_image():
288373
289374
290375@pytest .fixture
291- def linode_backup_enabled ():
376+ def linode_backup_enabled (linode_cloud_firewall ):
292377 # create linode with backups enabled
293378 linode_id = (
294379 exec_test_command (
@@ -306,6 +391,8 @@ def linode_backup_enabled():
306391 DEFAULT_TEST_IMAGE ,
307392 "--root_pass" ,
308393 DEFAULT_RANDOM_PASS ,
394+ "--firewall_id" ,
395+ linode_cloud_firewall ,
309396 "--text" ,
310397 "--no-headers" ,
311398 "--format=id" ,
@@ -348,14 +435,16 @@ def snapshot_of_linode():
348435
349436# Test helpers specific to Nodebalancers test suite
350437@pytest .fixture
351- def nodebalancer_with_default_conf ():
438+ def nodebalancer_with_default_conf (linode_cloud_firewall ):
352439 result = (
353440 exec_test_command (
354441 NODEBALANCER_BASE_CMD
355442 + [
356443 "create" ,
357444 "--region" ,
358445 "us-ord" ,
446+ "--firewall_id" ,
447+ linode_cloud_firewall ,
359448 "--text" ,
360449 "--delimiter" ,
361450 "," ,
@@ -452,10 +541,11 @@ def pytest_configure(config):
452541
453542
454543@pytest .fixture
455- def created_linode_id ( ):
544+ def support_test_linode_id ( linode_cloud_firewall ):
456545 timestamp = str (time .time_ns ())
457546 label = "cli" + timestamp
458- result = (
547+
548+ res = (
459549 exec_test_command (
460550 LINODE_BASE_CMD
461551 + [
@@ -470,20 +560,23 @@ def created_linode_id():
470560 label ,
471561 "--root_pass" ,
472562 DEFAULT_RANDOM_PASS ,
563+ "--firewall_id" ,
564+ linode_cloud_firewall ,
473565 "--text" ,
474566 "--delimiter" ,
475567 "," ,
476568 "--no-headers" ,
477569 "--format" ,
478- "label,region,type,image, id" ,
570+ "id" ,
479571 "--no-defaults" ,
480572 ]
481573 )
482574 .stdout .decode ()
483575 .rstrip ()
484576 )
485577
486- res_arr = result . split ( "," )
487- linode_id = res_arr [ 4 ]
578+ linode_id = res
579+
488580 yield linode_id
581+
489582 delete_target_id (target = "linodes" , id = linode_id )
0 commit comments