Skip to content

Commit

Permalink
Merge branch 'feature/#214-update-zia-traffic-forwarding-endpoints' i…
Browse files Browse the repository at this point in the history
…nto develop
  • Loading branch information
mitchos committed Sep 14, 2023
2 parents 66a640a + 7909e7e commit b7ab615
Show file tree
Hide file tree
Showing 5 changed files with 283 additions and 5 deletions.
2 changes: 2 additions & 0 deletions pyzscaler/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@ def snake_to_camel(name: str):
"surrogate_ip": "surrogateIP",
"surrogate_ip_enforced_for_known_browsers": "surrogateIPEnforcedForKnownBrowsers",
"ec_vms": "ecVMs",
"ipv6_enabled": "ipV6Enabled",
}
return edge_cases.get(name, name[0].lower() + name.title()[1:].replace("_", ""))

Expand All @@ -31,6 +32,7 @@ def camel_to_snake(name: str):
"surrogateIP": "surrogate_ip",
"surrogateIPEnforcedForKnownBrowsers": "surrogate_ip_enforced_for_known_browsers",
"ecVMs": "ec_vms",
"ipV6Enabled": "ipv6_enabled",
}
# Check if name is an edge case
if name in edge_cases:
Expand Down
70 changes: 70 additions & 0 deletions pyzscaler/zia/locations.py
Original file line number Diff line number Diff line change
Expand Up @@ -421,3 +421,73 @@ def delete_location(self, location_id: str) -> int:
"""
return self._delete(f"locations/{location_id}", box=False).status_code

def get_geo_by_coordinates(self, latitude: int, longitude: int) -> Box:
"""
Retrieves the geographical data of the region or city that is located in the specified latitude and longitude
coordinates. The geographical data includes the city name, state, country, geographical ID of the city and
state, etc.
Args:
latitude (int): The latitude of the location.
longitude (int): The longitude of the location.
Returns:
:obj:`Box`: The geographical data of the region or city that is located in the specified coordinates.
Examples:
Get the geographical data of the region or city that is located in the specified coordinates::
print(zia.locations.get_geo_by_coordinates(37.3860517, -122.0838511))
"""
payload = {"latitude": latitude, "longitude": longitude}
return self._get("region/byGeoCoordinates", params=payload)

def get_geo_by_ip(self, ip: str) -> Box:
"""
Retrieves the geographical data of the region or city that is located in the specified IP address. The
geographical data includes the city name, state, country, geographical ID of the city and state, etc.
Args:
ip (str): The IP address of the location.
Returns:
:obj:`Box`: The geographical data of the region or city that is located in the specified IP address.
Examples:
Get the geographical data of the region or city that is located in the specified IP address::
print(zia.locations.get_geo_by_ip("8.8.8.8")
"""
return self._get(f"region/byIPAddress/{ip}")

def list_cities_by_name(self, **kwargs) -> BoxList:
"""
Retrieves the list of cities (along with their geographical data) that match the prefix search. The geographical
data includes the latitude and longitude coordinates of the city, geographical ID of the city and state,
country, postal code, etc.
Args:
**kwargs: Optional keyword arguments.
Keyword Args:
prefix (str): The prefix string to search for cities.
page (int): The page number of the results.
page_size (int): The number of results per page.
Returns:
:obj:`BoxList`: The list of cities (along with their geographical data) that match the prefix search.
Examples:
Get the list of cities (along with their geographical data) that match the prefix search::
for city in zia.locations.list_cities_by_name(prefix="San Jose"):
print(city)
Notes:
Very broad or generic search terms may return a large number of results which can take a long time to be
returned. Ensure you narrow your search result as much as possible to avoid this.
"""
return BoxList(Iterator(self._api, "region/search", **kwargs))
79 changes: 79 additions & 0 deletions pyzscaler/zia/traffic.py
Original file line number Diff line number Diff line change
Expand Up @@ -661,3 +661,82 @@ def delete_vpn_credential(self, credential_id: str) -> int:
"""
return self._delete(f"vpnCredentials/{credential_id}", box=False).status_code

def get_ipv6_config(self) -> Box:
"""
Returns the IPv6 configuration for the organisation.
Returns:
:obj:`Box`: The IPv6 configuration for the organisation.
Examples:
Get the IPv6 configuration for the organisation::
zia.traffic.get_ipv6_config()
"""
# There is an edge case in the camelcase to snake conversion that we're going to handle here. We'll revert
# the default camel_killer_box and run it through our conversion function in utils that handles edge-cases.
self._box_attrs = {"camel_killer_box": False}

return convert_keys(self._get("ipv6config"), "to_snake")

def list_dns64_prefixes(self, **kwargs):
"""
Returns the list of NAT64 prefixes configured as the DNS64 prefix for the organisation
Keyword Args:
search (str): Search string to filter results by. Defaults to None.
Returns:
:obj:`BoxList`: List of NAT64 prefixes configured as the DNS64 prefix for the organisation
Examples:
List DNS64 prefixes using default settings::
for prefix in zia.traffic.list_dns64_prefixes():
print(prefix)
"""

return self._get("ipv6config/dns64prefix", params=kwargs)

def list_nat64_prefixes(self, **kwargs) -> BoxList:
"""
Returns the list of NAT64 prefixes configured for the organisation
Keyword Args:
page (int): Page number to return. Defaults to 1.
page_size (int): Number of results to return per page. Defaults to 100. Max size is 1000.
search (str, optional): Search string to filter results by. Defaults to None.
Returns:
:obj:`BoxList`: List of NAT64 prefixes configured for the organisation
Examples:
List NAT64 prefixes using default settings::
for prefix in zia.traffic.list_nat64_prefixes():
print(prefix)
"""
return BoxList(Iterator(self._api, "ipv6config/nat64prefix", **kwargs))

def list_gre_ip_addresses(self, **kwargs) -> BoxList:
"""
Returns a list of IP addresses with GRE tunnel details.
Keyword Args:
ip_addresses (list[str]): Filter based on the list of IP addresses provided.
Returns:
:obj:`BoxList`: List of GRE IP addresses configured for the organisation
Examples:
List GRE IP addresses using default settings::
for ip_address in zia.traffic.list_gre_ip_addresses():
print(ip_address)
"""
return self._get("orgProvisioning/ipGreTunnelInfo", params=convert_keys(kwargs))
74 changes: 71 additions & 3 deletions tests/zia/test_locations.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
import pytest
import responses
from box import Box
from box import Box, BoxList
from responses import matchers

from tests.conftest import stub_sleep
Expand Down Expand Up @@ -66,6 +66,22 @@ def fixture_sub_locations():
]


@pytest.fixture(name="geo_locations")
def geo_location_data():
return {
"city_geo_id": 5375480,
"state_geo_id": 5332921,
"latitude": 37.3897,
"longitude": -122.0832,
"city_name": "Mountain View",
"state_name": "California",
"country_name": "United States",
"country_code": "US",
"postal_code": "94041",
"continent_code": "NA",
}


@responses.activate
@stub_sleep
def test_list_locations_with_one_page(zia, paginated_items):
Expand Down Expand Up @@ -183,7 +199,7 @@ def test_get_location_by_id(zia, locations):
def test_get_location_by_name_and_id(zia):
# Passing location_id and location_name should result in a ValueError.
with pytest.raises(ValueError):
resp = zia.locations.get_location(location_id="1", location_name="Test A")
zia.locations.get_location(location_id="1", location_name="Test A")


@responses.activate
Expand Down Expand Up @@ -267,7 +283,7 @@ def test_update_location(zia, locations):

resp = zia.locations.update_location("1", name="Updated Test")

assert isinstance(resp, dict)
assert isinstance(resp, Box)
assert resp.id == 1
assert resp.name == "Updated Test"

Expand Down Expand Up @@ -381,3 +397,55 @@ def test_list_sublocations(zia, sub_locations):
assert isinstance(resp, list)
assert len(resp) == 2
assert resp[0].id == 1


@responses.activate
def test_get_geo_by_coordinates(zia, geo_locations):
responses.add(
responses.GET,
url="https://zsapi.zscaler.net/api/v1/region/byGeoCoordinates",
json=geo_locations,
status=200,
)
resp = zia.locations.get_geo_by_coordinates(37, -122)
assert isinstance(resp, Box)
assert resp.city_name == "Mountain View"


@responses.activate
def test_get_geo_by_ip(zia, geo_locations):
responses.add(
responses.GET,
url="https://zsapi.zscaler.net/api/v1/region/byIPAddress/8.8.8.8",
json=geo_locations,
status=200,
)
resp = zia.locations.get_geo_by_ip("8.8.8.8")
assert isinstance(resp, Box)
assert resp.city_name == "Mountain View"


@stub_sleep
@responses.activate
def test_list_cities_by_name(zia):
list_cities_data = [
{"city": "San Jose", "state": "CA", "country": "US"},
{"city": "San Francisco", "state": "CA", "country": "US"},
]

responses.add(
responses.GET,
url="https://zsapi.zscaler.net/api/v1/region/search?prefix=San&page=1",
json=list_cities_data,
status=200,
)
responses.add(
responses.GET,
url="https://zsapi.zscaler.net/api/v1/region/search?prefix=San&page=2",
json=[],
status=200,
)
resp = zia.locations.list_cities_by_name(prefix="San")
assert isinstance(resp, BoxList)
assert len(resp) == 2
assert resp[0].city == "San Jose"
63 changes: 61 additions & 2 deletions tests/zia/test_traffic.py
Original file line number Diff line number Diff line change
Expand Up @@ -215,6 +215,14 @@ def fixture_vpn_credentials():
]


@pytest.fixture(name="ipv6_prefixes")
def fixture_ipv64_prefixes():
return [
{"id": 0, "name": "sample1", "prefixMask": "mask1", "dnsPrefix": True},
{"id": 1, "name": "sample2", "prefixMask": "mask2", "dnsPrefix": False},
]


@responses.activate
@stub_sleep
def test_list_gre_tunnels(zia, gre_tunnels):
Expand Down Expand Up @@ -508,8 +516,8 @@ def test_get_vpn_credential_by_fqdn(zia, vpn_credentials):


def test_get_vpn_credential_error(zia):
with pytest.raises(Exception) as e_info:
resp = zia.traffic.get_vpn_credential("1", "[email protected]")
with pytest.raises(Exception):
zia.traffic.get_vpn_credential("1", "[email protected]")


@responses.activate
Expand Down Expand Up @@ -575,3 +583,54 @@ def test_list_vips(zia, vips):
assert isinstance(resp, BoxList)
assert len(resp) == 2
assert resp[0].data_center == "TESTA"


@responses.activate
def test_list_dns64_prefixes(zia, ipv6_prefixes):
responses.add(
responses.GET,
url="https://zsapi.zscaler.net/api/v1/ipv6config/dns64prefix",
json=ipv6_prefixes,
status=200,
)
resp = zia.traffic.list_dns64_prefixes()
assert isinstance(resp, BoxList)
assert len(resp) == 2


@stub_sleep
@responses.activate
def test_list_nat64_prefixes(zia, ipv6_prefixes):
responses.add(
responses.GET,
url="https://zsapi.zscaler.net/api/v1/ipv6config/nat64prefix?page=1",
json=ipv6_prefixes,
status=200,
)
responses.add(
responses.GET,
url="https://zsapi.zscaler.net/api/v1/ipv6config/nat64prefix?page=2",
json=[],
status=200,
)
resp = zia.traffic.list_nat64_prefixes()
assert isinstance(resp, BoxList)
assert len(resp) == 2


@responses.activate
def test_list_gre_ip_addresses(zia):
gre_ip_addresses_data = [
{"ipAddress": "192.168.1.1", "greEnabled": True, "greTunnelIP": "10.0.0.1"},
{"ipAddress": "192.168.1.2", "greEnabled": False, "greTunnelIP": "10.0.0.2"},
]

responses.add(
responses.GET,
url="https://zsapi.zscaler.net/api/v1/orgProvisioning/ipGreTunnelInfo",
json=gre_ip_addresses_data,
status=200,
)
resp = zia.traffic.list_gre_ip_addresses()
assert isinstance(resp, BoxList)
assert len(resp) == 2

0 comments on commit b7ab615

Please sign in to comment.