Skip to content

ref: remove duplicate code #64

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
amarao merged 2 commits into from
Jun 15, 2026
Merged

ref: remove duplicate code #64

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
db03872
ref: remove duplicate code
amarao May 26, 2026
006bb51
ref: remove code duplication for _find_l2_segment_id
amarao May 26, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
133 changes: 66 additions & 67 deletions ansible_collections/serverscom/sc_api/plugins/module_utils/cloud_computing.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,51 @@
__metaclass__ = type


def _wait_for_instance_status(
api,
instance,
instance_id,
wait,
update_interval,
status_done,
statuses_continue,
initial_sleep=False,
):
start_time = time.time()
if wait and initial_sleep:
time.sleep(update_interval)
while instance["status"] not in statuses_continue + [status_done]:
if not wait:
break
if time.time() > start_time + wait:
raise WaitError(
msg=f"Timeout waiting instance {instance['id']} "
f"status {status_done} or {statuses_continue}. "
f"Last state was {instance['status']}",
timeout=time.time() - start_time,
)
time.sleep(update_interval)
instance = api.get_instances(
instance_id,
retry_rules=_retry_rules_for_wait(
max_wait=max(0, wait - (time.time() - start_time)),
delay=update_interval,
),
)
if instance["status"] == status_done:
return instance, True
if instance["status"] in statuses_continue:
return instance, False
if wait:
raise WaitError(
msg=f"Timeout waiting instance {instance['id']} "
f"status {status_done}. "
f"Last state was {instance['status']}",
timeout=time.time() - start_time,
)
return instance, None


class ScCloudComputingRegionsInfo(object):
def __init__(self, endpoint, token, search_pattern):
self.search_pattern = search_pattern
Expand Down Expand Up @@ -447,38 +492,16 @@ def __init__(
self.checkmode = checkmode

def wait_for_statuses(self, status_done, statuses_continue):
start_time = time.time()
while self.instance["status"] not in statuses_continue + [status_done]:
if not self.wait:
break
if time.time() > start_time + self.wait:
raise WaitError(
msg=f"Timeout waiting instance {self.instance['id']} "
f"status {status_done} or {statuses_continue}. "
f"Last state was {self.instance['status']}",
timeout=time.time() - start_time,
)
time.sleep(self.update_interval)
self.instance = self.api.get_instances(
self.instance_id,
retry_rules=_retry_rules_for_wait(
max_wait=max(0, self.wait - (time.time() - start_time)),
delay=self.update_interval,
),
)
if self.instance["status"] == status_done:
return True
else:
if self.instance["status"] in statuses_continue:
return False
else:
if self.wait:
raise WaitError(
msg=f"Timeout waiting instance {self.instance['id']} "
f"status {status_done}. "
f"Last state was {self.instance['status']}",
timeout=time.time() - start_time,
)
self.instance, ready = _wait_for_instance_status(
api=self.api,
instance=self.instance,
instance_id=self.instance_id,
wait=self.wait,
update_interval=self.update_interval,
status_done=status_done,
statuses_continue=statuses_continue,
)
return ready

def shutdown(self):
if self.instance["status"] == "RESCUE":
Expand Down Expand Up @@ -593,42 +616,18 @@ def __init__(
self.update_interval = update_interval
self.checkmode = checkmode

# copypaste, refactor, TODO
def wait_for_statuses(self, status_done, statuses_continue):
start_time = time.time()
if self.wait:
time.sleep(self.update_interval) # workaround around bug in APIs
while self.instance["status"] not in statuses_continue + [status_done]:
if not self.wait:
break
if time.time() > start_time + self.wait:
raise WaitError(
msg=f"Timeout waiting instance {self.instance['id']} "
f"status {status_done} or {statuses_continue}. "
f"Last state was {self.instance['status']}",
timeout=time.time() - start_time,
)
time.sleep(self.update_interval)
self.instance = self.api.get_instances(
self.instance["id"],
retry_rules=_retry_rules_for_wait(
max_wait=max(0, self.wait - (time.time() - start_time)),
delay=self.update_interval,
),
)
if self.instance["status"] == status_done:
return True
else:
if self.instance["status"] in statuses_continue:
return False
else:
if self.wait:
raise WaitError(
msg=f"Timeout waiting instance {self.instance['id']} "
f"status {status_done}. "
f"Last state was {self.instance['status']}",
timeout=time.time() - start_time,
)
self.instance, ready = _wait_for_instance_status(
api=self.api,
instance=self.instance,
instance_id=self.instance["id"],
wait=self.wait,
update_interval=self.update_interval,
status_done=status_done,
statuses_continue=statuses_continue,
initial_sleep=True, # workaround around bug in APIs
)
return ready

def run(self):
if self.checkmode:
Expand Down
60 changes: 26 additions & 34 deletions ansible_collections/serverscom/sc_api/plugins/module_utils/l2_segment.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,26 @@
__metaclass__ = type


def _find_l2_segment_id(api, segment_id=None, name=None, type=None, must=False):
existing_segment_id = None
if segment_id:
if must:
return api.get_l2_segment(segment_id)["id"]
return api.get_l2_segment_or_none(segment_id)["id"]

for segment in api.list_l2_segments():
name_matches = segment["name"] == name
type_matches = not type or segment["type"] == type
if name_matches and type_matches:
if existing_segment_id:
raise ModuleError(msg=f"Duplicate segment with name {name} found.")
existing_segment_id = segment["id"]

if must and not existing_segment_id:
raise ModuleError(f"Segment {name} is not found.")
return existing_segment_id


class ScL2SegmentsInfo:
def __init__(self, endpoint, token, label_selector):
self.api = ScApi(token, endpoint)
Expand Down Expand Up @@ -86,26 +106,10 @@ def __init__(
if update_interval > wait:
raise ModuleError("update_interval is longer than wait")

@staticmethod
def _match_segment(api_object, segment_name, type):
if type:
return api_object["name"] == segment_name and api_object["type"] == type
else:
return api_object["name"] == segment_name

def get_segment_id(self):
existing_segment_id = None
if self.segment_id:
existing_segment_id = self.api.get_l2_segment_or_none(self.segment_id)["id"]
else:
for segment in self.api.list_l2_segments():
if self._match_segment(segment, self.name, self.type):
if existing_segment_id: # duplicate found
raise ModuleError(
msg=f"Duplicate segment with name {self.name} found."
)
existing_segment_id = segment["id"]
return existing_segment_id
return _find_l2_segment_id(
self.api, segment_id=self.segment_id, name=self.name, type=self.type
)

def wait_for_active_segment(self, segment_id):
ready = False
Expand Down Expand Up @@ -347,22 +351,10 @@ def __init__(
self.update_interval = update_interval
self.checkmode = checkmode

# TODO: code repeated from ScL2Segment
def get_segment_id(self):
existing_segment_id = None
if self.segment_id:
existing_segment_id = self.api.get_l2_segment(self.segment_id)["id"]
else:
for segment in self.api.list_l2_segments():
if segment["name"] == self.name:
if existing_segment_id: # duplicate found
raise ModuleError(
msg=f"Duplicate segment with name {self.name} found."
)
existing_segment_id = segment["id"]
if not existing_segment_id:
raise ModuleError(f"Segment {self.name} is not found.")
return existing_segment_id
return _find_l2_segment_id(
self.api, segment_id=self.segment_id, name=self.name, must=True
)

def wait_for(self, l2):
start_time = time.time()
Expand Down
39 changes: 39 additions & 0 deletions ansible_collections/serverscom/sc_api/tests/unit/plugins/modules/test_l2_segment.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,8 +3,14 @@
# (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)

from __future__ import absolute_import, division, print_function
import pytest
import mock
from ansible_collections.serverscom.sc_api.plugins.module_utils.modules import (
ModuleError,
)
from ansible_collections.serverscom.sc_api.plugins.module_utils.l2_segment import (
ScL2Segment,
_find_l2_segment_id,
) # noqa


Expand Down Expand Up @@ -56,3 +62,36 @@ def test_simplify_members():
}
]
assert list(ScL2Segment._simplify_members(data)) == data


def test_find_l2_segment_id_matches_type():
api = mock.MagicMock()
api.list_l2_segments.return_value = [
{"id": "l2-1", "name": "shared", "type": "sbm"},
{"id": "l2-2", "name": "shared", "type": "dedicated"},
]

assert _find_l2_segment_id(api, name="shared", type="dedicated") == "l2-2"


def test_find_l2_segment_id_raises_when_required_segment_missing():
api = mock.MagicMock()
api.list_l2_segments.return_value = []

with pytest.raises(ModuleError) as exc_info:
_find_l2_segment_id(api, name="missing", must=True)

assert "Segment missing is not found" in exc_info.value.msg


def test_find_l2_segment_id_raises_on_duplicate_name():
api = mock.MagicMock()
api.list_l2_segments.return_value = [
{"id": "l2-1", "name": "shared", "type": "sbm"},
{"id": "l2-2", "name": "shared", "type": "sbm"},
]

with pytest.raises(ModuleError) as exc_info:
_find_l2_segment_id(api, name="shared")

assert "Duplicate segment with name shared found" in exc_info.value.msg