Skip to content

Commit

Permalink
Friendlier interface for remote configuration job creation
Browse files Browse the repository at this point in the history
  • Loading branch information
@vduseev
vduseev committed Dec 6, 2024
1 parent 5fd98f3 commit 6840a05
Show file tree
Hide file tree
Showing 3 changed files with 74 additions and 100 deletions.
46 changes: 32 additions & 14 deletions dynatrace/environment_v2/remote_configuration.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -45,12 +45,14 @@ def list(self, time_from: Optional[Union[datetime, str]] = None, time_to: Option
target_params=params
)

def post(self, payload: "RemoteConfigurationManagementOperationActiveGateRequest") -> "RemoteConfigurationManagementJob":
def post(self, entities: List[str], operations: List["RemoteConfigurationManagementOperation"]) -> "RemoteConfigurationManagementJob":
"""Creates a new remote configuration management job
:param payload: The remote configuration management job definition
:param entities: The list of entities to apply the operations to
:param operations: The list of operations to apply
:return: The created job
"""
payload = RemoteConfigurationManagementOperationActiveGateRequest(entities=entities, operations=operations)
response = self.__http_client.make_request(
self.ENDPOINT,
method="POST",
Expand All @@ -68,12 +70,14 @@ def get_current(self) -> Optional["RemoteConfigurationManagementJob"]:
return None
return RemoteConfigurationManagementJob(raw_element=response.json())

def post_preview(self, payload: "RemoteConfigurationManagementOperationActiveGateRequest") -> "PaginatedList[RemoteConfigurationManagementJobPreview]":
def post_preview(self, entities: List[str], operations: List["RemoteConfigurationManagementOperation"]) -> "PaginatedList[RemoteConfigurationManagementJobPreview]":
"""Creates a preview for remote configuration management job
:param payload: The remote configuration management job definition
:param entities: The list of entities to apply the operations to
:param operations: The list of operations to apply
:return: A paginated list of job previews
"""
payload = RemoteConfigurationManagementOperationActiveGateRequest(entities=entities, operations=operations)
return PaginatedList(
RemoteConfigurationManagementJobPreview,
self.__http_client,
Expand All @@ -82,18 +86,19 @@ def post_preview(self, payload: "RemoteConfigurationManagementOperationActiveGat
target_params=payload.to_json()
)

def validate(self, payload: "RemoteConfigurationManagementOperationActiveGateRequest") -> Optional["RemoteConfigurationManagementValidationResult"]:
def validate(self, entities: List[str], operations: List["RemoteConfigurationManagementOperation"]) -> Optional["RemoteConfigurationManagementValidationResult"]:
"""Validates the payload for a remote configuration management job
:param payload: The remote configuration management job definition to validate
:param entities: The list of entities to apply the operations to
:param operations: The list of operations to apply
:return: Validation result if validation failed, None if validation succeeded
"""
payload = RemoteConfigurationManagementOperationActiveGateRequest(entities=entities, operations=operations)
response = self.__http_client.make_request(
f"{self.ENDPOINT}/validator",
method="POST",
data=payload.to_json()
)
print(f"RESPONSE: '{response.content}'")
if not response.content:
return None
return RemoteConfigurationManagementValidationResult(raw_element=response.json())
Expand Down Expand Up @@ -132,12 +137,14 @@ def list(self, time_from: Optional[Union[datetime, str]] = None, time_to: Option
target_params=params
)

def post(self, payload: "RemoteConfigurationManagementOperationOneAgentRequest") -> "RemoteConfigurationManagementJob":
def post(self, entities: List[str], operations: List["RemoteConfigurationManagementOperation"]) -> "RemoteConfigurationManagementJob":
"""Creates a new remote configuration management job
:param payload: The remote configuration management job definition
:param entities: The list of entities to apply the operations to
:param operations: The list of operations to apply
:return: The created job
"""
payload = RemoteConfigurationManagementOperationOneAgentRequest(entities=entities, operations=operations)
response = self.__http_client.make_request(
self.ENDPOINT,
method="POST",
Expand All @@ -155,12 +162,14 @@ def get_current(self) -> Optional["RemoteConfigurationManagementJob"]:
return None
return RemoteConfigurationManagementJob(raw_element=response.json())

def post_preview(self, payload: "RemoteConfigurationManagementOperationOneAgentRequest") -> "PaginatedList[RemoteConfigurationManagementJobPreview]":
def post_preview(self, entities: List[str], operations: List["RemoteConfigurationManagementOperation"]) -> "PaginatedList[RemoteConfigurationManagementJobPreview]":
"""Creates a preview for remote configuration management job
:param payload: The remote configuration management job definition
:param entities: The list of entities to apply the operations to
:param operations: The list of operations to apply
:return: A paginated list of job previews
"""
payload = RemoteConfigurationManagementOperationOneAgentRequest(entities=entities, operations=operations)
return PaginatedList(
RemoteConfigurationManagementJobPreview,
self.__http_client,
Expand All @@ -169,18 +178,19 @@ def post_preview(self, payload: "RemoteConfigurationManagementOperationOneAgentR
target_params=payload.to_json()
)

def validate(self, payload: "RemoteConfigurationManagementOperationOneAgentRequest") -> Optional["RemoteConfigurationManagementValidationResult"]:
def validate(self, entities: List[str], operations: List["RemoteConfigurationManagementOperation"]) -> Optional["RemoteConfigurationManagementValidationResult"]:
"""Validates the payload for a remote configuration management job
:param payload: The remote configuration management job definition to validate
:param entities: The list of entities to apply the operations to
:param operations: The list of operations to apply
:return: Validation result if validation failed, None if validation succeeded
"""
payload = RemoteConfigurationManagementOperationOneAgentRequest(entities=entities, operations=operations)
response = self.__http_client.make_request(
f"{self.ENDPOINT}/validator",
method="POST",
data=payload.to_json()
)
print(f"RESPONSE: '{response.content}'")
if not response.content:
return None
return RemoteConfigurationManagementValidationResult(raw_element=response.json())
Expand Down Expand Up @@ -228,6 +238,14 @@ def to_json(self) -> Dict[str, Any]:
"operation": self.operation.value,
"value": self.value
}

@staticmethod
def build(attribute: AttributeType, operation: OperationType, value: Optional[str] = None) -> "RemoteConfigurationManagementOperation":
return RemoteConfigurationManagementOperation(raw_element={
"attribute": attribute.value,
"operation": operation.value,
"value": value
})

class RemoteConfigurationManagementOperationActiveGateRequest:
"""Remote configuration management operation creation request"""
Expand Down
66 changes: 21 additions & 45 deletions test/environment_v2/test_activegates_remote_configuration.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -28,18 +28,13 @@ def test_list(dt: Dynatrace):
break

def test_post(dt: Dynatrace):
operation = RemoteConfigurationManagementOperation(raw_element={
"attribute": "networkZone",
"operation": "set",
"value": "test-zone"
})

request = RemoteConfigurationManagementOperationActiveGateRequest(
entities=[TEST_ENTITY_ID],
operations=[operation]
operation = RemoteConfigurationManagementOperation.build(
attribute=AttributeType.NETWORK_ZONE,
operation=OperationType.SET,
value="test-zone",
)

job = dt.activegates_remote_configuration.post(request)
job = dt.activegates_remote_configuration.post(entities=[TEST_ENTITY_ID], operations=[operation])

assert job is not None
assert job.id is not None
Expand All @@ -58,19 +53,17 @@ def test_get_current(dt: Dynatrace):
assert current_job.processed_entities_count <= current_job.total_entities_count

def test_post_preview(dt: Dynatrace):
operation = RemoteConfigurationManagementOperation(raw_element={
"attribute": "networkZone",
"operation": "set",
"value": "test-zone"
})
operation = RemoteConfigurationManagementOperation.build(
attribute=AttributeType.NETWORK_ZONE,
operation=OperationType.SET,
value="test-zone",
)

request = RemoteConfigurationManagementOperationActiveGateRequest(
previews = dt.activegates_remote_configuration.post_preview(
entities=[TEST_ENTITY_ID],
operations=[operation]
operations=[operation],
)

previews = dt.activegates_remote_configuration.post_preview(request)

assert isinstance(previews, PaginatedList)

for preview in previews:
Expand All @@ -83,19 +76,16 @@ def test_post_preview(dt: Dynatrace):
break

def test_validate(dt: Dynatrace):
operation = RemoteConfigurationManagementOperation(raw_element={
"attribute": "networkZone",
"operation": "set",
"value": "test-zone"
})

request = RemoteConfigurationManagementOperationActiveGateRequest(
operation = RemoteConfigurationManagementOperation.build(
attribute=AttributeType.NETWORK_ZONE,
operation=OperationType.SET,
value="test-zone",
)
validation_result = dt.activegates_remote_configuration.validate(
entities=[TEST_ENTITY_ID],
operations=[operation]
)

validation_result = dt.activegates_remote_configuration.validate(request)

# If validation succeeds, result should be None
# If validation fails, result should contain error details
if validation_result is not None:
Expand All @@ -105,25 +95,11 @@ def test_validate(dt: Dynatrace):
assert isinstance(validation_result.invalid_operations, list)

def test_get_job(dt: Dynatrace):
# First create a job to get its ID
operation = RemoteConfigurationManagementOperation(raw_element={
"attribute": "networkZone",
"operation": "set",
"value": "test-zone"
})

request = RemoteConfigurationManagementOperationActiveGateRequest(
entities=[TEST_ENTITY_ID],
operations=[operation]
)

created_job = dt.activegates_remote_configuration.post(request)

# Then get the job by ID
job = dt.activegates_remote_configuration.get(created_job.id)
ID = "7974003406714390819"
job = dt.activegates_remote_configuration.get(ID)

assert job is not None
assert job.id == created_job.id
assert job.id == ID
assert job.entity_type == EntityType.ACTIVE_GATE
assert len(job.operations) == 1
assert job.operations[0].attribute == AttributeType.NETWORK_ZONE
Expand Down
62 changes: 21 additions & 41 deletions test/environment_v2/test_oneagents_remote_configuration.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -28,19 +28,17 @@ def test_list(dt: Dynatrace):
break

def test_post(dt: Dynatrace):
operation = RemoteConfigurationManagementOperation(raw_element={
"attribute": "networkZone",
"operation": "set",
"value": "test-zone"
})
operation = RemoteConfigurationManagementOperation.build(
attribute=AttributeType.NETWORK_ZONE,
operation=OperationType.SET,
value="test-zone"
)

request = RemoteConfigurationManagementOperationOneAgentRequest(
job = dt.oneagents_remote_configuration.post(
entities=[TEST_ENTITY_ID],
operations=[operation]
)

job = dt.oneagents_remote_configuration.post(request)

assert job is not None
assert job.id is not None
assert job.entity_type == EntityType.ONE_AGENT
Expand All @@ -58,19 +56,17 @@ def test_get_current(dt: Dynatrace):
assert current_job.processed_entities_count <= current_job.total_entities_count

def test_post_preview(dt: Dynatrace):
operation = RemoteConfigurationManagementOperation(raw_element={
"attribute": "networkZone",
"operation": "set",
"value": "test-zone"
})
operation = RemoteConfigurationManagementOperation.build(
attribute=AttributeType.NETWORK_ZONE,
operation=OperationType.SET,
value="test-zone"
)

request = RemoteConfigurationManagementOperationOneAgentRequest(
previews = dt.oneagents_remote_configuration.post_preview(
entities=[TEST_ENTITY_ID],
operations=[operation]
)

previews = dt.oneagents_remote_configuration.post_preview(request)

assert isinstance(previews, PaginatedList)

for preview in previews:
Expand All @@ -83,19 +79,17 @@ def test_post_preview(dt: Dynatrace):
break

def test_validate(dt: Dynatrace):
operation = RemoteConfigurationManagementOperation(raw_element={
"attribute": "networkZone",
"operation": "set",
"value": "test-zone"
})
operation = RemoteConfigurationManagementOperation.build(
attribute=AttributeType.NETWORK_ZONE,
operation=OperationType.SET,
value="test-zone",
)

request = RemoteConfigurationManagementOperationOneAgentRequest(
validation_result = dt.oneagents_remote_configuration.validate(
entities=[TEST_ENTITY_ID],
operations=[operation]
)

validation_result = dt.oneagents_remote_configuration.validate(request)

# If validation succeeds, result should be None
# If validation fails, result should contain error details
if validation_result is not None:
Expand All @@ -105,25 +99,11 @@ def test_validate(dt: Dynatrace):
assert isinstance(validation_result.invalid_operations, list)

def test_get_job(dt: Dynatrace):
# First create a job to get its ID
operation = RemoteConfigurationManagementOperation(raw_element={
"attribute": "networkZone",
"operation": "set",
"value": "test-zone"
})

request = RemoteConfigurationManagementOperationOneAgentRequest(
entities=[TEST_ENTITY_ID],
operations=[operation]
)

created_job = dt.oneagents_remote_configuration.post(request)

# Then get the job by ID
job = dt.oneagents_remote_configuration.get(created_job.id)
ID = "7974003406714390819"
job = dt.oneagents_remote_configuration.get(ID)

assert job is not None
assert job.id == created_job.id
assert job.id == ID
assert job.entity_type == EntityType.ONE_AGENT
assert len(job.operations) == 1
assert job.operations[0].attribute == AttributeType.NETWORK_ZONE
Expand Down

0 comments on commit 6840a05

Please sign in to comment.