import rasterize
from rasterize import *
import demistomock as demisto
from CommonServerPython import entryTypes
from tempfile import NamedTemporaryFile
from pytest_mock import MockerFixture
from unittest.mock import MagicMock
import os
import logging
import http.server
import time
import threading
import pytest
import requests
import json
# disable warning from urllib3. these are emitted when python driver can't connect to chrome yet
logging.getLogger("urllib3").setLevel(logging.ERROR)
RETURN_ERROR_TARGET = "rasterize.return_error"
def util_read_tsv(file_path):
with open(file_path) as file:
ret_value = file.read()
return ret_value
def util_load_json(path):
with open(path, encoding="utf-8") as f:
return json.loads(f.read())
class MockPychromeEventHandler:
is_mailto = False
is_private_network_url = False
document_url = ""
path = ""
class MockTab:
id = "tab_id"
def test_rasterize_email_image(caplog, capfd, mocker):
with capfd.disabled() and NamedTemporaryFile("w+") as f:
f.write(
''
"
---------- TEST FILE ----------
"
)
path = os.path.realpath(f.name)
f.flush()
mocker.patch.object(rasterize, "support_multithreading")
perform_rasterize(path=f"file://{path}", width=250, height=250, rasterize_type=RasterizeType.PNG)
caplog.clear()
def test_rasterize_email_image_array(caplog, capfd, mocker):
mocker.patch("rasterize.demisto.command", return_value="rasterize-email")
with capfd.disabled() and NamedTemporaryFile("w+") as f:
f.write(
''
"
---------- TEST FILE ----------
"
)
path = os.path.realpath(f.name)
f.flush()
mocker.patch.object(rasterize, "support_multithreading")
perform_rasterize(path=[f"file://{path}"], width=250, height=250, rasterize_type=RasterizeType.PNG)
caplog.clear()
def test_rasterize_email_pdf(caplog, capfd, mocker):
mocker.patch("rasterize.demisto.command", return_value="rasterize-pdf")
with capfd.disabled() and NamedTemporaryFile("w+") as f:
f.write(
''
"
---------- TEST FILE ----------
"
)
path = os.path.realpath(f.name)
f.flush()
mocker.patch.object(rasterize, "support_multithreading")
perform_rasterize(path=f"file://{path}", width=250, height=250, rasterize_type=RasterizeType.PDF)
caplog.clear()
def test_get_chrome_options():
res = get_chrome_options(CHROME_OPTIONS, "")
assert res == CHROME_OPTIONS
res = get_chrome_options(CHROME_OPTIONS, "[--disable-dev-shm-usage],--disable-auto-reload, --headless")
assert "--disable-dev-shm-usage" not in res
assert "--no-sandbox" in res # part of default options
assert "--disable-auto-reload" in res
assert len([x for x in res if x == "--headless"]) == 1 # should have only one headless option
res = get_chrome_options(CHROME_OPTIONS, r"--user-agent=test\,comma")
assert len([x for x in res if x.startswith("--user-agent")]) == 1
assert "--user-agent=test,comma" in res
res = get_chrome_options(CHROME_OPTIONS, r"[--user-agent]") # remove user agent
assert len([x for x in res if x.startswith("--user-agent")]) == 0
def test_rasterize_large_html(capfd, mocker):
mocker.patch("rasterize.demisto.command", return_value="rasterize-html")
with capfd.disabled():
path = os.path.realpath("test_data/large.html")
mocker.patch.object(rasterize, "support_multithreading")
res = perform_rasterize(path=f"file://{path}", width=250, height=250, rasterize_type=RasterizeType.PNG)
assert res
def test_rasterize_html(mocker, capfd):
mocker.patch("rasterize.demisto.command", return_value="rasterize-html")
with capfd.disabled():
path = os.path.realpath("test_data/file.html")
mocker.patch.object(demisto, "args", return_value={"EntryID": "test"})
mocker.patch.object(demisto, "getFilePath", return_value={"path": path})
mocker.patch.object(os, "rename")
mocker.patch.object(os.path, "realpath", return_value=f"{os.getcwd()}/test_data/file.html")
mocker_output = mocker.patch("rasterize.return_results")
mocker.patch.object(rasterize, "support_multithreading")
rasterize_html_command()
assert mocker_output.call_args.args[0]["File"] == "email.png"
@pytest.fixture
def http_wait_server():
# Simple http handler which waits 10 seconds before responding
class WaitHanlder(http.server.BaseHTTPRequestHandler):
def do_HEAD(self):
self.send_response(200)
self.send_header("Content-type", "text/html")
self.end_headers()
def do_GET(self):
time.sleep(10)
try:
self.send_response(200)
self.send_header("Content-type", "text/html")
self.end_headers()
self.wfile.write(
bytes("Test wait handlerTest Wait
", "utf-8")
)
self.flush_headers()
except BrokenPipeError: # ignore broken pipe as socket might have been closed
pass
# disable logging
def log_message(self, format, *args):
pass
with http.server.ThreadingHTTPServer(("", 10888), WaitHanlder) as server:
server_thread = threading.Thread(target=server.serve_forever)
server_thread.start()
yield
server.shutdown()
server_thread.join()
# Some web servers can block the connection after the http is sent
# In this case chromium will hang. An example for this is:
# curl -v -H 'user-agent: HeadlessChrome' --max-time 10 "http://www.grainger.com/" # disable-secrets-detection
# This tests access a server which waits for 10 seconds and makes sure we timeout
@pytest.mark.filterwarnings("ignore::ResourceWarning")
def test_rasterize_url_long_load(mocker: MockerFixture, http_wait_server, capfd):
return_error_mock = mocker.patch(RETURN_ERROR_TARGET)
time.sleep(1) # give time to the server to start
with capfd.disabled():
mocker.patch.object(rasterize, "support_multithreading")
perform_rasterize("http://localhost:10888", width=250, height=250, rasterize_type=RasterizeType.PNG, navigation_timeout=5)
assert return_error_mock.call_count == 1
# call_args last call with a tuple of args list and kwargs
# err_msg = return_error_mock.call_args[0][0]
# assert 'Timeout exception' in err_msg
return_error_mock.reset_mock()
# test that with a higher value we get a response
assert perform_rasterize("http://localhost:10888", width=250, height=250, rasterize_type=RasterizeType.PNG)
assert not return_error_mock.called
@pytest.mark.filterwarnings("ignore::ResourceWarning")
def test_rasterize_image_to_pdf(mocker):
path = os.path.realpath("test_data/image.png")
mocker.patch.object(demisto, "args", return_value={"EntryID": "test"})
mocker.patch.object(demisto, "getFilePath", return_value={"path": path})
mocker.patch.object(demisto, "results")
mocker.patch.object(rasterize, "support_multithreading")
rasterize_image_command()
assert demisto.results.call_count == 1
# call_args is tuple (args list, kwargs). we only need the first one
results = demisto.results.call_args[0]
assert len(results) == 1
assert results[0][0]["Type"] == entryTypes["entryInfoFile"]
TEST_DATA = [
("test_data/many_pages.pdf", 21, 21, None),
("test_data/many_pages.pdf", 20, 20, None),
("test_data/many_pages.pdf", "*", 51, None),
(
"test_data/test_pw_mathias.pdf",
"*",
1,
"mathias",
),
]
@pytest.mark.parametrize("file_path, max_pages, expected_length, pw", TEST_DATA)
def test_convert_pdf_to_jpeg(file_path, max_pages, expected_length, pw):
res = convert_pdf_to_jpeg(file_path, max_pages, pw)
assert type(res) is list
assert len(res) == expected_length
@pytest.mark.parametrize(
"width, height, expected_width, expected_height", [(8001, 700, 8000, 700), (700, 80001, 700, 8000), (700, 600, 700, 600)]
)
def test_get_width_height(width, height, expected_width, expected_height):
"""
Given:
1. A width that is larger than the safeguard limit, and a valid height
2. A height that is larger than the safeguard limit, and a valid width
3. Valid width and height
When:
- Running the 'heck_width_and_height' function.
Then:
Verify that:
1. The resulted width is the safeguard limit (8000px) and the height remains as it was.
2. The resulted height is the safeguard limit (8000px) and the width remains as it was.
3. Both width and height remain as they were.
"""
args = {"width": str(width), "height": str(height)}
w, h = get_width_height(args)
assert w == expected_width
assert h == expected_height
class TestRasterizeIncludeUrl:
class MockChromeOptions:
def __init__(self) -> None:
self.options = []
def add_argument(self, arg):
self.options.append(arg)
class MockChrome:
def __init__(self, options, service) -> None:
self.options = options.options
self.page_source = ""
self.session_id = "session_id"
def set_page_load_timeout(self, max_page_load_time):
pass
def get(self, path):
pass
def maximize_window(self):
pass
def implicitly_wait(self, arg):
pass
def set_window_size(self, width, height):
pass
def get_screenshot_as_png(self):
return "image"
def quit(self):
pass
@pytest.mark.parametrize("include_url", [False, True])
def test_sanity_rasterize_with_include_url(self, mocker: MockerFixture, include_url: bool, capfd: pytest.CaptureFixture):
"""
Given:
- A parameter that mention whether to include the URL bar in the screenshot.
When:
- Running the 'rasterize' function.
Then:
- Verify that it runs as expected.
"""
mocker.patch("os.remove")
mocker.patch("rasterize.demisto.command", return_value="rasterize-image")
with capfd.disabled(), NamedTemporaryFile("w+") as f:
f.write(
''
"
---------- TEST FILE ----------
"
)
path = os.path.realpath(f.name)
f.flush()
mocker.patch.object(rasterize, "support_multithreading")
image = perform_rasterize(
path=f"file://{path}", width=250, height=250, rasterize_type=RasterizeType.PNG, include_url=include_url
)
assert image
def test_log_warning():
"""
Given pypdf's logger instance
When checking the logger's level.
Then make sure the level is ERROR
"""
import logging
from rasterize import pypdf_logger
assert pypdf_logger.level == logging.ERROR
assert pypdf_logger.level == logging.ERROR
def test_excepthook_recv_loop(mocker):
"""
Given Exceptions that might happen after the tab was closed.
When A chromium tab is closed.
Then make sure the right info is logged.
"""
mock_args = type("mock_args", (), dict.fromkeys(("exc_type", "exc_value")))
demisto_info = mocker.patch.object(demisto, "info")
excepthook_recv_loop(mock_args)
demisto_info.assert_any_call("Unsuppressed Exception in _recv_loop: args.exc_type=None")
demisto_info.assert_any_call("Unsuppressed Exception in _recv_loop: args.exc_type=None, empty exc_value")
def test_poppler_version():
import pdf2image
poppler_version = pdf2image.pdf2image._get_poppler_version("pdftoppm")
assert poppler_version[0] > 20
def test_get_list_item():
from rasterize import get_list_item
my_list = ["a", "b", "c"]
assert get_list_item(my_list, 0, "FOO") == "a"
assert get_list_item(my_list, 1, "FOO") == "b"
assert get_list_item(my_list, 2, "FOO") == "c"
assert get_list_item(my_list, 3, "FOO") == "FOO"
assert get_list_item(my_list, 4, "FOO") == "FOO"
def test_add_filename_suffix():
from rasterize import add_filename_suffix
my_list = ["a", "b", "c"]
my_list_with_suffix = add_filename_suffix(my_list, "sfx")
assert len(my_list) == len(my_list_with_suffix)
for current_element_index, _ in enumerate(my_list):
assert f"{my_list[current_element_index]}.sfx" == my_list_with_suffix[current_element_index]
def test_get_output_filenames():
from rasterize import get_list_item, add_filename_suffix
file_name = ["foo_01", "foo_02", "foo_03"]
file_names = argToList(file_name)
file_names = add_filename_suffix(file_names, "png")
assert get_list_item(file_names, 0, "FOO.png") == "foo_01.png"
assert get_list_item(file_names, 1, "FOO.png") == "foo_02.png"
assert get_list_item(file_names, 2, "FOO.png") == "foo_03.png"
assert get_list_item(file_names, 3, "FOO.png") == "FOO.png"
assert get_list_item(file_names, 4, "FOO.png") == "FOO.png"
def test_chrome_manager_case_chrome_instances_file_is_empty(mocker):
"""
Given instance id and chrome options
When chrome instances file is empty
Then make sure code running into case 1 calling generate_new_chrome_instance which return browser and chrome port.
"""
from rasterize import chrome_manager
instance_id = "new_instance_id"
chrome_options = "new_chrome_options"
mock_context = {"context": {"IntegrationInstanceID": instance_id}}
params = {"chrome_options": chrome_options}
mocker.patch.object(demisto, "callingContext", mock_context)
mocker.patch.object(demisto, "params", return_value=params)
mocker.patch.object(rasterize, "read_json_file", return_value={})
generate_new_chrome_instance_mocker = mocker.patch.object(
rasterize, "generate_new_chrome_instance", return_value=["browser_object", "chrome_port"]
)
terminate_chrome_mocker = mocker.patch.object(rasterize, "terminate_chrome", return_value=None)
browser, chrome_port = chrome_manager()
assert generate_new_chrome_instance_mocker.call_count == 1
generate_new_chrome_instance_mocker.assert_called_with(instance_id, chrome_options)
assert terminate_chrome_mocker.call_count == 0
assert browser == "browser_object"
assert chrome_port == "chrome_port"
def test_chrome_manager_case_chromes_options_exist_and_instance_id_not_linked(mocker):
"""
Given instance id that does not exist and chrome options that exist in the chrome instances file
When chrome instances file is not empty and instance id is not linked to the chrome options
Then make sure code running into case 2 and calling generate_new_chrome_instance which return browser and chrome port.
"""
from rasterize import chrome_manager, read_json_file
instance_id = "instance_id_that_does_not_exist"
chrome_options = "chrome_options2"
mock_context = {"context": {"IntegrationInstanceID": instance_id}}
params = {"chrome_options": chrome_options}
mock_file_content = read_json_file("test_data/chrome_instances.json")
mocker.patch.object(demisto, "callingContext", mock_context)
mocker.patch.object(demisto, "params", return_value=params)
mocker.patch.object(rasterize, "read_json_file", return_value=mock_file_content)
generate_new_chrome_instance_mocker = mocker.patch.object(
rasterize, "generate_new_chrome_instance", return_value=["browser_object", "chrome_port"]
)
terminate_chrome_mocker = mocker.patch.object(rasterize, "terminate_chrome", return_value=None)
browser, chrome_port = chrome_manager()
assert generate_new_chrome_instance_mocker.call_count == 1
generate_new_chrome_instance_mocker.assert_called_with(instance_id, chrome_options)
assert terminate_chrome_mocker.call_count == 0
assert browser == "browser_object"
assert chrome_port == "chrome_port"
def test_chrome_manager_case_new_chrome_options_and_instance_id(mocker):
"""
Given instance id and chrome options does not exist in the chrome instances file
When chrome instances file is not empty
Then make sure code running into case 3 and calling generate_new_chrome_instance which return browser and chrome port.
"""
from rasterize import chrome_manager, read_json_file
instance_id = "instance_id_that_does_not_exist"
chrome_options = "chrome_options_that_does_not_exist"
mock_context = {"context": {"IntegrationInstanceID": instance_id}}
params = {"chrome_options": chrome_options}
mock_file_content = read_json_file("test_data/chrome_instances.json")
mocker.patch.object(demisto, "callingContext", mock_context)
mocker.patch.object(demisto, "params", return_value=params)
mocker.patch.object(rasterize, "read_json_file", return_value=mock_file_content)
generate_new_chrome_instance_mocker = mocker.patch.object(
rasterize, "generate_new_chrome_instance", return_value=["browser_object", "chrome_port"]
)
terminate_chrome_mocker = mocker.patch.object(rasterize, "terminate_chrome", return_value=None)
browser, chrome_port = chrome_manager()
assert generate_new_chrome_instance_mocker.call_count == 1
generate_new_chrome_instance_mocker.assert_called_with(instance_id, chrome_options)
assert terminate_chrome_mocker.call_count == 0
assert browser == "browser_object"
assert chrome_port == "chrome_port"
def test_chrome_manager_case_instance_id_exist_but_new_chrome_options(mocker):
"""
Given instance id exist and chrome options does not exist in the chrome instances file
When chrome instances file is not empty and instance id has different chrome options
Then make sure code running into case 4, terminating old chrome port, generating new one,
and update the chrome instances file.
"""
from rasterize import chrome_manager, read_json_file
instance_id = "22222222-2222-2222-2222-222222222222" # exist
chrome_options = "chrome_options_that_does_not_exist"
mock_context = {"context": {"IntegrationInstanceID": instance_id}}
params = {"chrome_options": chrome_options}
mock_file_content = read_json_file("test_data/chrome_instances.json")
mocker.patch.object(demisto, "callingContext", mock_context)
mocker.patch.object(demisto, "params", return_value=params)
mocker.patch.object(rasterize, "read_json_file", return_value=mock_file_content)
mocker.patch.object(rasterize, "get_chrome_browser", return_value=None)
terminate_chrome_mocker = mocker.patch.object(rasterize, "terminate_chrome", return_value=None)
generate_new_chrome_instance_mocker = mocker.patch.object(
rasterize, "generate_new_chrome_instance", return_value=["browser_object", "chrome_port"]
)
browser, chrome_port = chrome_manager()
assert terminate_chrome_mocker.call_count == 1
assert generate_new_chrome_instance_mocker.call_count == 1
generate_new_chrome_instance_mocker.assert_called_with(instance_id, chrome_options)
assert browser == "browser_object"
assert chrome_port == "chrome_port"
def test_chrome_manager_case_instance_id_and_chrome_options_exist_and_linked(mocker):
"""
Given instance id and chrome options
When chrome instances file is not empty, and instance id and chrome options linked.
Then make sure code running into case 5 and using the browser that already in used.
"""
from rasterize import chrome_manager, read_json_file
instance_id = "22222222-2222-2222-2222-222222222222" # exist
chrome_options = "chrome_options2"
mock_context = {"context": {"IntegrationInstanceID": instance_id}}
params = {"chrome_options": chrome_options}
mock_file_content = read_json_file("test_data/chrome_instances.json")
mocker.patch.object(demisto, "callingContext", mock_context)
mocker.patch.object(demisto, "params", return_value=params)
mocker.patch.object(rasterize, "read_json_file", return_value=mock_file_content)
mocker.patch.object(rasterize, "get_chrome_browser", return_value="browser_object")
terminate_chrome_mocker = mocker.patch.object(rasterize, "terminate_chrome", return_value=None)
generate_new_chrome_instance_mocker = mocker.patch.object(
rasterize, "generate_new_chrome_instance", return_value=["browser_object", "chrome_port"]
)
browser, chrome_port = chrome_manager()
assert terminate_chrome_mocker.call_count == 0
assert generate_new_chrome_instance_mocker.call_count == 0
assert browser == "browser_object"
assert chrome_port == "2222"
def test_generate_chrome_port():
"""
Given first_chrome_port and max_chromes_count
When needed to generate new chrome port
Then make sure the function generate valid chrome port.
"""
from rasterize import generate_chrome_port
port = generate_chrome_port()
assert 0 <= len(port) <= 5
def test_generate_chrome_port_no_port_available(mocker):
"""
Given first_chrome_port and max_chromes_count that creates empty range
When needed to generate new chrome port
Then make sure the function will raise an error and return None
"""
from rasterize import generate_chrome_port
rasterize.FIRST_CHROME_PORT = 0
rasterize.MAX_CHROMES_COUNT = 0
mock_return_error = mocker.patch.object(demisto, "error", return_value=None)
port = generate_chrome_port()
assert mock_return_error.call_count == 1
assert not port
def test_get_chrome_browser_error(mocker: MockerFixture):
"""
Given A connection error.
When Launching a pychrome browser.
Then Make sure the error is caught and debugged properly.
"""
from rasterize import get_chrome_browser
def raise_connection_error(url):
raise requests.exceptions.ConnectionError("connection error")
mocker.patch.object(rasterize, "get_chrome_processes", return_value=["/bin/bash /opt/google/chrome/google-chrome"])
mocker.patch("pychrome.Browser", side_effect=raise_connection_error)
mocker.patch("time.sleep")
debug = mocker.patch.object(demisto, "debug")
res = get_chrome_browser("port")
assert res is None
debug.assert_called_with(
"Failed to connect to Chrome on port port on iteration 4. ConnectionError,"
" exp_str='connection error', exp=ConnectionError('connection error')"
)
def test_backoff(mocker):
"""
Given Waiting for a process to complete.
When Launching a pychrome browser.
Then Make sure to wait the required amount.
"""
from rasterize import backoff
sleep_mock = mocker.patch("time.sleep")
res = backoff(None, 2, 1)
assert res == (None, 2)
sleep_mock.assert_called_with(1)
def test_is_mailto_urls(mocker: MockerFixture):
"""
Given A mailto URL is called.
When Attempting to make a screenshot.
Then Make sure the correct output is returned.
"""
from rasterize import screenshot_image
mock_handler = MockPychromeEventHandler()
mock_handler.is_mailto = True
mocker.patch("rasterize.navigate_to_path", return_value=mock_handler)
res = screenshot_image(None, MockTab(), "url", None, None)
assert res == (None, 'URLs that start with "mailto:" cannot be rasterized.\nURL: url')
def test_get_frame_tree_url_success(mocker: MockerFixture):
"""
Test the get_frame_tree_url method of PychromeEventHandler with successful retrieval.
Given:
- A PychromeEventHandler object with a mock tab that successfully returns a frame tree
When:
- Calling the get_frame_tree_url method
Then:
- Verify the correct URL is returned
- Verify the tab_ready_event is not set
"""
mock_tab = mocker.Mock()
mock_tab.id = "test_tab_id"
mock_tab.Page.getFrameTree.return_value = {"frameTree": {"frame": {"url": "https://test.com"}}}
mock_event = mocker.Mock()
handler = PychromeEventHandler(None, mock_tab, mock_event, "https://test.com", 30)
assert handler.get_frame_tree_url() == "https://test.com"
assert not mock_event.set.called
def test_screenshot_image_local_file(mocker: MockerFixture):
"""The function returns an error when attempting to rasterize a local file"""
mock_browser = mocker.Mock()
mock_tab = mocker.Mock()
mocker.patch("rasterize.demisto.command", return_value="rasterize-test")
local_file_path = "file:///path/to/local/file.html"
result, error_message = screenshot_image(
mock_browser,
mock_tab,
local_file_path,
wait_time=0,
navigation_timeout=30,
full_screen=False,
include_url=False,
include_source=False,
)
assert result is None
assert error_message == "Cannot rasterize local files"
def test_is_private_network_urls(mocker: MockerFixture):
"""
Given A private network URL is called.
When Attempting to make a screenshot.
Then Make sure the correct output is returned.
"""
from rasterize import screenshot_image
mock_handler = MockPychromeEventHandler()
mock_handler.is_private_network_url = True
mocker.patch("rasterize.navigate_to_path", return_value=mock_handler)
res = screenshot_image(None, MockTab(), "url", None, None)
assert res == (
None,
'URLs that belong to the "This" Network (0.0.0.0/8),'
" or the Loopback Network (127.0.0.0/8) cannot be rasterized.\nURL: url",
)
def test_increase_counter_chrome_instances_file(mocker):
"""
Given:
- A new Chrome instance content
- A valid Chrome port
- An increase counter
- A terminate port
When:
- Executing the increase_counter_chrome_instances_file function
Then:
- The function writes to the correct file and increase the "RASTERIZATION_COUNT" by 1
"""
from rasterize import increase_counter_chrome_instances_file, RASTERIZATION_COUNT
from unittest.mock import mock_open
mocker.patch("os.path.exists", return_value=True)
mock_file_content = util_load_json("test_data/chrome_instances.json")
expected_rasterization_count = mock_file_content["2222"][RASTERIZATION_COUNT] + 1
mock_file = mock_open()
mocker.patch("builtins.open", mock_file)
mocker.patch.object(json, "load", return_value=mock_file_content)
mocker_json = mocker.patch("json.dump")
increase_counter_chrome_instances_file(chrome_port="2222")
assert mocker_json.called
assert expected_rasterization_count == mocker_json.call_args[0][0]["2222"][RASTERIZATION_COUNT]
def test_add_new_chrome_instance(mocker):
"""
Given:
- A new Chrome instance content
When:
- Executing the add_new_chrome_instance function
Then:
- The function writes to the correct file the new chrome instance.
"""
from rasterize import add_new_chrome_instance
from unittest.mock import mock_open
mocker.patch("os.path.exists", return_value=True)
mock_file_content = util_load_json("test_data/chrome_instances.json")
mock_file = mock_open()
mocker.patch("builtins.open", mock_file)
mocker.patch.object(json, "load", return_value=mock_file_content)
mocker_json = mocker.patch("json.dump")
add_new_chrome_instance(
new_chrome_instance_content={
"9345": {
"instance_id": "44444444-4444-4444-4444-444444444444",
"chrome_options": "chrome_options4",
"rasterize_count": 1,
}
}
)
assert mocker_json.called
assert "9345" in mocker_json.call_args[0][0]
def test_terminate_port_chrome_instances_file(mocker):
"""
Given:
- A port to terminate.
When:
- Executing the terminate_port_chrome_instances_file function
Then:
- The function writes to the correct file the data without the port to terminate.
"""
from rasterize import terminate_port_chrome_instances_file
from unittest.mock import mock_open
mocker.patch("os.path.exists", return_value=True)
mock_file_content = util_load_json("test_data/chrome_instances.json")
mock_file = mock_open()
mocker.patch("builtins.open", mock_file)
mocker.patch.object(json, "load", return_value=mock_file_content)
mocker_json = mocker.patch("json.dump")
terminate_port_chrome_instances_file(chrome_port="2222")
assert mocker_json.called
assert "2222" not in mocker_json.call_args[0][0]
def test_write_chrome_instances_empty(mocker):
"""
Given:
- A new Chrome instance content(first chrome instance).
When:
- Executing the write_chrome_instances_file function
Then:
- The function creates and writes to the correct file, calls json.dump with the expected arguments.
"""
from rasterize import write_chrome_instances_file
from unittest.mock import mock_open
mock_file_content = util_load_json("test_data/chrome_instances.json")
mock_file = mock_open()
mocker.patch("builtins.open", mock_file)
mocker_json = mocker.patch.object(json, "dump", return_value=mock_file_content)
write_chrome_instances_file(new_chrome_content=mock_file_content)
assert mocker_json.call_count == 1
def test_read_json_file(mocker):
"""
Given:
- A JSON file at 'test_data/chrome_instances.json'
When:
- Executing the read_json_file function
Then:
- The function reads the JSON file and returns the correct content.
"""
from rasterize import read_json_file
mocker.patch("os.path.exists", return_value=True)
mock_file_content = util_load_json("test_data/chrome_instances.json")
file_result = read_json_file("test_data/chrome_instances.json")
assert file_result == mock_file_content
def test_rasterize_mailto(capfd, mocker):
"""
Given:
- mailto argument as path.
When:
- Running the 'rasterize' function.
Then:
- Verify that perform_rasterize exit with the expected error message.
"""
mocker_output = mocker.patch("rasterize.return_results")
with pytest.raises(SystemExit) as excinfo, capfd.disabled():
perform_rasterize(path="mailto:some.person@gmail.com", width=250, height=250, rasterize_type=RasterizeType.PNG)
assert mocker_output.call_args.args[0].readable_output == (
"The following paths were skipped as they are not valid for rasterization: ['mailto:some.person@gmail.com']"
)
assert excinfo.type is SystemExit
assert excinfo.value.code == 0
def test_rasterize_private_network(capfd: pytest.CaptureFixture, mocker: MockerFixture):
"""
Given:
- argument as path.
When:
- Running the 'rasterize' function.
Then:
- Verify that perform_rasterize exit with the expected error message.
"""
mocker_output = mocker.patch("rasterize.return_results")
with pytest.raises(SystemExit) as excinfo, capfd.disabled():
perform_rasterize(path="0.0.0.8/test", width=250, height=250, rasterize_type=RasterizeType.PNG)
assert mocker_output.call_args.args[0].readable_output == (
"The following paths were skipped as they are not valid for rasterization: ['0.0.0.8/test']"
)
assert excinfo.type is SystemExit
assert excinfo.value.code == 0
@pytest.mark.parametrize(
"url, expected",
[
pytest.param("http://192.168.1.1", True, id="private IPv4"),
pytest.param("https://10.0.0.1", True, id="private IPv4 with HTTPS"),
pytest.param("localhost", False, id="localhost"),
pytest.param("http://8.8.8.8", False, id="public IPv4"),
pytest.param("invalid_url", False, id="invalid URL"),
pytest.param("http://", False, id="empty URL"),
pytest.param("192.168.1.1", True, id="private IPv4"),
pytest.param("2001:db8::1", False, id="IPv6 address"),
pytest.param("https://www.example.com", False, id="public domain"),
pytest.param("http://127.0.0.1", True, id="loopback IPv4"),
pytest.param("http://0.0.0.1", True, id="this network IPv4"),
],
)
def test_is_private_network(url: str, expected: bool):
assert is_private_network(url) == expected
@pytest.mark.parametrize(
"test_input,expected",
[
pytest.param("192.168.001.001", "192.168.1.1", id="ipv4_with_leading_zeros"),
pytest.param("010.001.001.001", "10.1.1.1", id="ipv4_with_leading_zeros_all_octets"),
pytest.param("http://192.168.001.001", "http://192.168.1.1", id="url_with_ipv4_leading_zeros"),
pytest.param("example.com", "example.com", id="domain_name_unchanged"),
pytest.param("256.1.2.3", "256.1.2.3", id="invalid_ipv4_unchanged"),
pytest.param("", "", id="empty_string"),
pytest.param("127.000.000.001", "127.0.0.1", id="localhost_with_leading_zeros"),
pytest.param("127.000.000.001:8080", "127.0.0.1:8080", id="localhost_with_leading_zeros_and_port"),
pytest.param("http://192.168.001.001?param=value", "http://192.168.1.1?param=value", id="url_with_ipv4_and_query_params"),
pytest.param("2001:db8:3333:4444:5555:6666:7777:8888", "2001:db8:3333:4444:5555:6666:7777:8888", id="ipv6"),
],
)
def test_remove_leading_zeros_from_ip_addresses(test_input: str, expected: str):
"""
Test the remove_leading_zeros_from_ip_addresses function with various inputs.
"""
assert remove_leading_zeros_from_ip_addresses(test_input) == expected
def test_handle_request_paused(mocker: MockerFixture):
"""
Given:
- cloudflare.com as BLOCKED_URLS parameter.
When:
- Running the 'handle_request_paused' function.
Then:
- Verify that tab.Fetch.failRequest executed with the correct requestId and errorReason Aborted
"""
mocker.patch("rasterize.BLOCKED_URLS", ["cloudflare.com"])
kwargs = {"requestId": "1", "request": {"url": "cloudflare.com"}}
mock_tab = MagicMock(spec=pychrome.Tab)
mock_fetch = mocker.MagicMock()
mock_fetch.disable = MagicMock()
mock_fail_request = mocker.patch.object(mock_fetch, "failRequest", new_callable=MagicMock)
mock_tab.Fetch = mock_fetch
mock_tab.id = "mock_tab_id"
tab_event_handler = PychromeEventHandler(None, mock_tab, None, "", 0)
tab_event_handler.handle_request_paused(**kwargs)
assert mock_fail_request.call_args[1]["requestId"] == "1"
assert mock_fail_request.call_args[1]["errorReason"] == "Aborted"
def test_handle_request_paused_continues_non_blocked_url(mocker: MockerFixture):
"""
Given:
- cloudflare.com as BLOCKED_URLS parameter.
- A paused request for a URL that does NOT match any blocked pattern.
When:
- Running the 'handle_request_paused' function.
Then:
- Verify that tab.Fetch.continueRequest is called with the correct requestId,
and tab.Fetch.failRequest is NOT called.
This is the defensive `else` branch that prevents non-blocked paused requests
from hanging forever (the bug that caused the 180s page load deadlock).
"""
mocker.patch("rasterize.BLOCKED_URLS", ["cloudflare.com"])
kwargs = {"requestId": "2", "request": {"url": "https://example.com/main.css"}}
mock_tab = MagicMock(spec=pychrome.Tab)
mock_fetch = mocker.MagicMock()
mock_fail_request = mocker.patch.object(mock_fetch, "failRequest", new_callable=MagicMock)
mock_continue_request = mocker.patch.object(mock_fetch, "continueRequest", new_callable=MagicMock)
mock_tab.Fetch = mock_fetch
mock_tab.id = "mock_tab_id"
tab_event_handler = PychromeEventHandler(None, mock_tab, None, "", 0)
tab_event_handler.handle_request_paused(**kwargs)
mock_fail_request.assert_not_called()
assert mock_continue_request.call_args[1]["requestId"] == "2"
def test_handle_request_paused_blocks_multiple_matching_urls(mocker: MockerFixture):
"""
Given:
- cloudflare.com as BLOCKED_URLS parameter.
- Multiple sequential paused requests for different URLs that all match the blocked pattern.
When:
- Running 'handle_request_paused' for each of them.
Then:
- Verify that tab.Fetch.failRequest is called for EVERY matching URL (not just the first).
This proves the pre-fix bug where `Fetch.disable()` was called after the first abort,
which caused subsequent matching URLs to silently slip through, is closed.
"""
mocker.patch("rasterize.BLOCKED_URLS", ["cloudflare.com"])
mock_tab = MagicMock(spec=pychrome.Tab)
mock_fetch = mocker.MagicMock()
mock_fail_request = mocker.patch.object(mock_fetch, "failRequest", new_callable=MagicMock)
mock_tab.Fetch = mock_fetch
mock_tab.id = "mock_tab_id"
tab_event_handler = PychromeEventHandler(None, mock_tab, None, "", 0)
blocked_requests = [
{"requestId": "10", "request": {"url": "https://cdnjs.cloudflare.com/lib/a.css"}},
{"requestId": "11", "request": {"url": "https://cdnjs.cloudflare.com/lib/b.js"}},
{"requestId": "12", "request": {"url": "https://cdnjs.cloudflare.com/lib/c.js"}},
]
for kwargs in blocked_requests:
tab_event_handler.handle_request_paused(**kwargs)
assert mock_fail_request.call_count == 3
aborted_request_ids = [call.kwargs["requestId"] for call in mock_fail_request.call_args_list]
assert aborted_request_ids == ["10", "11", "12"]
def test_tab_lifecycle_manager_enables_fetch_with_scoped_patterns(mocker: MockerFixture):
"""
Given:
- BLOCKED_URLS contains two patterns: 'cloudflare.com' and 'tracker.io'.
When:
- TabLifecycleManager.__enter__ is invoked (the proactive Fetch.enable code path).
Then:
- Verify tab.Fetch.enable is called with a `patterns` argument containing exactly
one entry per blocked URL, each using glob-equivalent substring matching and
targeting the 'Request' stage.
This proves the proactive scoped-pattern fix is in place, preventing the race
condition where the first blocked URL slipped through unblocked and the
deadlock where bare Fetch.enable() intercepted (and hung) every request.
"""
blocked_urls = ["cloudflare.com", "tracker.io"]
mocker.patch("rasterize.BLOCKED_URLS", blocked_urls)
mock_tab = MagicMock(spec=pychrome.Tab)
mock_tab.Page = mocker.MagicMock()
mock_tab.Network = mocker.MagicMock()
mock_tab.Fetch = mocker.MagicMock()
mock_browser = mocker.MagicMock()
mock_browser.new_tab = mocker.MagicMock(return_value=mock_tab)
manager = rasterize.TabLifecycleManager(mock_browser, chrome_port=9301, offline_mode=False)
returned_tab = manager.__enter__()
assert returned_tab is mock_tab
mock_tab.Fetch.enable.assert_called_once()
call_kwargs = mock_tab.Fetch.enable.call_args.kwargs
assert "patterns" in call_kwargs
patterns = call_kwargs["patterns"]
assert len(patterns) == len(blocked_urls)
for pat, blocked in zip(patterns, blocked_urls):
assert pat["urlPattern"] == f"*{blocked}*"
assert pat["requestStage"] == "Request"
def test_tab_lifecycle_manager_skips_fetch_enable_when_no_blocked_urls(mocker: MockerFixture):
"""
Given:
- BLOCKED_URLS is empty (the common case for users not using URL blocking).
When:
- TabLifecycleManager.__enter__ is invoked.
Then:
- Verify tab.Fetch.enable is NOT called, so the integration doesn't pay any
interception cost when no URLs are configured to be blocked.
"""
mocker.patch("rasterize.BLOCKED_URLS", [])
mock_tab = MagicMock(spec=pychrome.Tab)
mock_tab.Page = mocker.MagicMock()
mock_tab.Network = mocker.MagicMock()
mock_tab.Fetch = mocker.MagicMock()
mock_browser = mocker.MagicMock()
mock_browser.new_tab = mocker.MagicMock(return_value=mock_tab)
manager = rasterize.TabLifecycleManager(mock_browser, chrome_port=9301, offline_mode=False)
manager.__enter__()
mock_tab.Fetch.enable.assert_not_called()
def test_retry_loading(mocker: MockerFixture):
"""
Test the retry_loading method of PychromeEventHandler
"""
mock_tab = mocker.Mock()
mock_tab.Page.navigate = mocker.Mock()
mock_tab.Page.getFrameTree = mocker.Mock(return_value={"frameTree": {"frame": {"url": CHROME_ERROR_URL}}})
mock_event = mocker.Mock()
handler = PychromeEventHandler(None, mock_tab, mock_event, "file:///test.html", 30)
mocker.patch("time.sleep")
# Test retry attempts failed
handler.retry_loading()
assert mock_tab.Page.navigate.call_count == 4
assert mock_tab.Page.getFrameTree.call_count == DEFAULT_RETRIES_COUNT
assert mock_event.set.called
# Test successful retry
mock_tab.Page.getFrameTree.return_value = {"frameTree": {"frame": {"url": "file:///test.html"}}}
handler.retry_loading()
assert mock_event.set.called
# Test getFrameTree failed to return URL
mock_tab.Page.getFrameTree.side_effect = pychrome.exceptions.RuntimeException("runtime_exception")
handler.retry_loading()
assert mock_event.set.called
@pytest.mark.parametrize(
"exception_class, exception_message",
[
pytest.param(pychrome.exceptions.RuntimeException, "Tab is closed", id="runtime_exception"),
pytest.param(pychrome.exceptions.UserAbortException, "User aborted", id="userabort_exception"),
],
)
def test_get_frame_tree_url_exception_handling(exception_class, exception_message, mocker: MockerFixture):
"""
Test the get_frame_tree_url method of PychromeEventHandler with exceptions.
Given:
- A PychromeEventHandler object with a mock tab that raises exceptions
When:
- Calling the get_frame_tree_url method
Then:
- Verify an empty string is returned
"""
mock_tab = mocker.Mock()
mock_tab.id = "test_tab_id"
mock_tab.Page.getFrameTree.side_effect = exception_class(exception_message)
mock_event = mocker.Mock()
handler = PychromeEventHandler(None, mock_tab, mock_event, "https://test.com", 30)
assert handler.get_frame_tree_url() == ""
@pytest.mark.parametrize(
"url, mock_event_set_called, mock_retry_loading_called",
[
pytest.param("http://test.com", True, False, id="http_url"),
pytest.param("file:///test.html", False, True, id="local_file"),
],
)
def test_page_frame_stopped_loading(
url: str, mock_event_set_called: bool, mock_retry_loading_called: bool, mocker: MockerFixture
):
"""
Test the page_frame_stopped_loading method of PychromeEventHandler for normal operation.
This test covers two scenarios:
1. HTTP URL: Verifies that the event is set when a regular page is loaded.
2. Local file: Checks if retry_loading is called when a local file is loaded.
Args:
url (str): The URL or file path to test.
mock_event_set_called (bool): Expected state of the event.set() call.
mock_retry_loading_called (bool): Expected state of the retry_loading() call.
mocker (MockerFixture): pytest-mock fixture for creating mock objects.
"""
mock_tab = mocker.Mock()
mock_event = mocker.Mock()
mock_tab.Page.getFrameTree.return_value = {"frameTree": {"frame": {"url": CHROME_ERROR_URL}}}
handler = PychromeEventHandler(None, mock_tab, mock_event, url, 30)
handler.start_frame = "test_frame_id"
mock_retry_loading = mocker.patch.object(handler, "retry_loading")
handler.page_frame_stopped_loading("test_frame_id")
assert mock_event.set.called == mock_event_set_called
assert mock_retry_loading.called == mock_retry_loading_called
@pytest.mark.parametrize(
"exception_class, exception_message",
[
pytest.param(pychrome.exceptions.RuntimeException, "Tab has been stopped", id="runtime_exception"),
pytest.param(pychrome.exceptions.UserAbortException, "User abort", id="userabort_exception"),
],
)
def test_page_frame_stopped_loading_exception_handling(exception_class, exception_message, mocker: MockerFixture):
"""
Test the page_frame_stopped_loading method of PychromeEventHandler for exception handling.
This test covers exception scenarios:
1. RuntimeException: When the tab has been stopped
2. UserAbortException: When the tab operation is aborted
Args:
exception_class: The exception class to simulate
exception_message: The message for the exception
mocker (MockerFixture): pytest-mock fixture for creating mock objects.
The test verifies that when exceptions occur, the event is properly set
to prevent hanging and no retry attempt is made.
"""
mock_tab = mocker.Mock()
mock_event = mocker.Mock()
mock_tab.Page.getFrameTree.side_effect = exception_class(exception_message)
handler = PychromeEventHandler(None, mock_tab, mock_event, "http://test.com", 30)
handler.start_frame = "test_frame_id"
mock_retry_loading = mocker.patch.object(handler, "retry_loading")
handler.page_frame_stopped_loading("test_frame_id")
assert mock_event.set.called
assert not mock_retry_loading.called
def test_chrome_manager_one_port_use_same_port(mocker):
"""
Given:
- instance id and chrome options.
When:
- Executing the chrome_manager_one_port function
Then:
- The function writes to the correct file the data and selects a port that already use the given chrome_option.
"""
from rasterize import chrome_manager_one_port, read_json_file
instance_id = "22222222-2222-2222-2222-222222222221" # not exist
chrome_options = "chrome_options2"
mock_context = {"context": {"IntegrationInstanceID": instance_id}}
params = {"chrome_options": chrome_options}
mock_file_content = read_json_file("test_data/chrome_instances.json")
mocker.patch.object(demisto, "callingContext", mock_context)
mocker.patch.object(demisto, "params", return_value=params)
mocker.patch.object(rasterize, "read_json_file", return_value=mock_file_content)
mocker.patch.object(rasterize, "get_chrome_browser", return_value="browser_object")
browser, chrome_port = chrome_manager_one_port()
assert browser == "browser_object"
assert chrome_port == "2222"
def test_chrome_manager_one_port_open_new_port(mocker):
"""
Given:
- instance id and chrome options.
When:
- Executing the chrome_manager_one_port function
Then:
- The function terminate all the ports that are open in chrome_manager, and opens a new chrome port to use.
"""
from rasterize import chrome_manager_one_port, read_json_file
instance_id = "22222222-2222-2222-2222-222222222221" # not exist
chrome_options = "new_chrome_options"
mock_context = {"context": {"IntegrationInstanceID": instance_id}}
params = {"chrome_options": chrome_options}
mock_file_content = read_json_file("test_data/chrome_instances.json")
mocker.patch.object(demisto, "callingContext", mock_context)
mocker.patch.object(demisto, "params", return_value=params)
mocker.patch.object(rasterize, "read_json_file", return_value=mock_file_content)
mocker.patch.object(rasterize, "get_chrome_browser", return_value="browser_object")
terminate_chrome_mocker = mocker.patch.object(rasterize, "terminate_chrome", return_value=None)
generate_new_chrome_instance_mocker = mocker.patch.object(
rasterize, "generate_new_chrome_instance", return_value=["browser_object", "chrome_port"]
)
browser, chrome_port = chrome_manager_one_port()
assert terminate_chrome_mocker.call_count == 3
assert generate_new_chrome_instance_mocker.call_count == 1
assert browser == "browser_object"
assert chrome_port == "chrome_port"
def test_rasterize_email_command_default_arge(mocker):
"""
Given: A valid HTML email body
When: The rasterize_email_command function is called
Then: The function should generate a PNG (default) image and return it as a file result
"""
from rasterize import rasterize_email_command
mock_args = {
"htmlBody": "Test email body
",
"width": "1000px",
"height": "1500px",
}
mocker.patch.object(demisto, "args", return_value=mock_args)
mock_perform_rasterize = mocker.patch("rasterize.perform_rasterize", return_value=[("image_data", None)])
mock_file_result = mocker.patch("rasterize.fileResult", return_value={"Type": "image"})
mock_uuid = mocker.patch("rasterize.uuid.uuid4", return_value="abcd-1234")
mocker.patch.object(demisto, "results")
rasterize_email_command()
mock_file_result.assert_called_once_with(filename=f"{mock_uuid.return_value}.png", data="image_data")
mock_perform_rasterize.assert_called_once_with(
path=mocker.ANY,
rasterize_type=RasterizeType.PNG,
width=1000,
height=1500,
offline_mode=False,
navigation_timeout=180,
full_screen=False,
)
def test_rasterize_email_command_png(mocker):
"""
Given: A valid HTML email body and PNG output type
When: The rasterize_email_command function is called
Then: The function should generate a PNG image and return it as a file result
"""
from rasterize import rasterize_email_command
mock_args = {"htmlBody": "Test email body
", "width": "800", "height": "600", "file_name": "test_email"}
mocker.patch.object(demisto, "args", return_value=mock_args)
mocker.patch("rasterize.perform_rasterize", return_value=[("image_data", None)])
mock_file_result = mocker.patch("rasterize.fileResult", return_value={"Type": "image"})
mock_results = mocker.patch.object(demisto, "results")
rasterize_email_command()
mock_file_result.assert_called_once_with(filename="test_email.png", data="image_data")
mock_results.assert_called_once()
def test_rasterize_email_command_pdf(mocker):
"""
Given: A valid HTML email body and PDF output type
When: The rasterize_email_command function is called
Then: The function should generate a PDF file and return it as a file result
"""
from rasterize import rasterize_email_command
mock_args = {"htmlBody": "Test email body
", "width": "800", "height": "600", "type": "pdf", "file_name": "test_email"}
mocker.patch.object(demisto, "args", return_value=mock_args)
mocker.patch("rasterize.perform_rasterize", return_value=[("pdf_data", None)])
mock_file_result = mocker.patch("rasterize.fileResult", return_value={"Type": "file"})
mock_results = mocker.patch.object(demisto, "results")
rasterize_email_command()
mock_file_result.assert_called_once_with(filename="test_email.pdf", data="pdf_data")
mock_results.assert_called_once()
def test_rasterize_email_command_full_screen(mocker):
"""
Given: A valid HTML email body and full_screen option set to true
When: The rasterize_email_command function is called
Then: The perform_rasterize function should be called with full_screen=True
"""
from rasterize import rasterize_email_command
mock_args = {"htmlBody": "Test email body
", "full_screen": "true", "type": "png", "file_name": "test_email"}
mocker.patch.object(demisto, "args", return_value=mock_args)
mock_perform_rasterize = mocker.patch("rasterize.perform_rasterize", return_value=[("image_data", None)])
mock_file_result = mocker.patch("rasterize.fileResult", return_value={"Type": "image"})
mocker.patch.object(demisto, "results")
rasterize_email_command()
mock_file_result.assert_called_once_with(filename="test_email.png", data="image_data")
mock_perform_rasterize.assert_called_once_with(
path=mocker.ANY,
rasterize_type=mocker.ANY,
width=mocker.ANY,
height=mocker.ANY,
offline_mode=mocker.ANY,
navigation_timeout=mocker.ANY,
full_screen=True,
)
def test_rasterize_email_command_offline_mode(mocker):
"""
Given: A valid HTML email body and offline mode set to true
When: The rasterize_email_command function is called
Then: The perform_rasterize function should be called with offline_mode=True
"""
from rasterize import rasterize_email_command
mock_args = {"htmlBody": "Test email body
", "offline": "true", "type": "png", "file_name": "test_email"}
mocker.patch.object(demisto, "args", return_value=mock_args)
mock_perform_rasterize = mocker.patch("rasterize.perform_rasterize", return_value=[("image_data", None)])
mocker.patch("rasterize.fileResult", return_value={"Type": "image"})
mocker.patch.object(demisto, "results")
rasterize_email_command()
mock_perform_rasterize.assert_called_once_with(
path=mocker.ANY,
rasterize_type=mocker.ANY,
width=mocker.ANY,
height=mocker.ANY,
offline_mode=True,
navigation_timeout=mocker.ANY,
full_screen=mocker.ANY,
)
def test_rasterize_email_command_custom_navigation_timeout(mocker):
"""
Given: A valid HTML email body and a custom navigation timeout
When: The rasterize_email_command function is called
Then: The perform_rasterize function should be called with the specified navigation_timeout
"""
from rasterize import rasterize_email_command
mock_args = {"htmlBody": "Test email body
", "max_page_load_time": "30", "type": "png", "file_name": "test_email"}
mocker.patch.object(demisto, "args", return_value=mock_args)
mock_perform_rasterize = mocker.patch("rasterize.perform_rasterize", return_value=[("image_data", None)])
mocker.patch("rasterize.fileResult", return_value={"Type": "image"})
mocker.patch.object(demisto, "results")
rasterize_email_command()
mock_perform_rasterize.assert_called_once_with(
path=mocker.ANY,
rasterize_type=mocker.ANY,
width=mocker.ANY,
height=mocker.ANY,
offline_mode=mocker.ANY,
navigation_timeout=30,
full_screen=mocker.ANY,
)
def test_rasterize_email_command_error_handling(mocker):
"""
Given: A scenario where perform_rasterize raises an exception
When: The rasterize_email_command function is called
Then: The function should log the error and return an error message
"""
from rasterize import rasterize_email_command
mock_args = {"htmlBody": "Test email body
", "type": "png", "file_name": "test_email"}
mocker.patch.object(demisto, "args", return_value=mock_args)
mocker.patch("rasterize.perform_rasterize", side_effect=Exception("Test error"))
mock_error = mocker.patch.object(demisto, "error")
with pytest.raises(SystemExit):
rasterize_email_command()
mock_error.assert_called_once_with("Test error")
def test_extract_content_from_tab_html(mocker):
"""
Given: A tab with HTML content
When: Calling extract_content_from_tab
Then: Should return markdown-formatted content and final URL
"""
from rasterize import extract_content_from_tab
mock_tab = mocker.Mock()
mock_tab.id = "test_tab_id"
mock_tab.Page.getFrameTree.return_value = {"frameTree": {"frame": {"url": "https://example.com"}}}
mock_tab.Runtime.evaluate.return_value = {
"result": {"value": {"type": "html", "content": "# Test Content\n\nThis is a test."}}
}
content, url = extract_content_from_tab(mock_tab, 30)
assert content == "# Test Content\n\nThis is a test."
assert url == "https://example.com"
mock_tab.Runtime.evaluate.assert_called_once()
def test_extract_content_from_tab_json(mocker):
"""
Given: A tab with JSON content
When: Calling extract_content_from_tab
Then: Should return formatted JSON and final URL
"""
from rasterize import extract_content_from_tab
mock_tab = mocker.Mock()
mock_tab.id = "test_tab_id"
mock_tab.Page.getFrameTree.return_value = {"frameTree": {"frame": {"url": "https://api.example.com/data"}}}
mock_tab.Runtime.evaluate.return_value = {"result": {"value": {"type": "json", "content": '{"key": "value", "number": 123}'}}}
content, url = extract_content_from_tab(mock_tab, 30)
assert '"key": "value"' in content
assert '"number": 123' in content
assert url == "https://api.example.com/data"
def test_extract_content_from_tab_empty_content(mocker, capfd):
"""
Given: A tab that returns empty content
When: Calling extract_content_from_tab
Then: Should raise DemistoException
"""
from rasterize import extract_content_from_tab
mock_tab = mocker.Mock()
mock_tab.id = "test_tab_id"
mock_tab.Page.getFrameTree.return_value = {"frameTree": {"frame": {"url": "https://example.com"}}}
mock_tab.Runtime.evaluate.return_value = {"result": {"value": {"type": "html", "content": ""}}}
with capfd.disabled():
content, url = extract_content_from_tab(mock_tab, 30)
assert content.startswith("Extraction Error:")
assert url == "https://example.com"
def test_extract_content_from_tab_exception(mocker, capfd):
"""
Given: A tab that raises an exception during extraction
When: Calling extract_content_from_tab
Then: Should return error message
"""
from rasterize import extract_content_from_tab
mock_tab = mocker.Mock()
mock_tab.id = "test_tab_id"
mock_tab.Page.getFrameTree.return_value = {"frameTree": {"frame": {"url": "https://example.com"}}}
mock_tab.Runtime.evaluate.side_effect = Exception("Test error")
with capfd.disabled():
content, url = extract_content_from_tab(mock_tab, 30)
assert "Extraction Error:" in content
assert "Test error" in content
assert url == "https://example.com"
def test_extract_text_content_success(mocker, capfd):
"""
Given: A valid URL for text extraction
When: Calling extract_text_content
Then: Should return extracted content and final URL
"""
from rasterize import extract_text_content
mock_browser = mocker.Mock()
mock_tab = mocker.Mock()
mock_tab.id = "test_tab_id"
mock_handler = mocker.Mock()
mock_handler.is_mailto = False
mock_handler.is_private_network_url = False
mocker.patch("rasterize.navigate_to_path", return_value=mock_handler)
mocker.patch("rasterize.extract_content_from_tab", return_value=("# Test Content", "https://example.com"))
with capfd.disabled():
content, url = extract_text_content(mock_browser, mock_tab, "https://example.com", 0, 30)
assert content == "# Test Content"
assert url == "https://example.com"
def test_extract_text_content_mailto_url(mocker, capfd):
"""
Given: A mailto URL
When: Calling extract_text_content
Then: Should return None and error message
"""
from rasterize import extract_text_content
mock_browser = mocker.Mock()
mock_tab = mocker.Mock()
mock_tab.id = "test_tab_id"
mock_handler = mocker.Mock()
mock_handler.is_mailto = True
mock_handler.is_private_network_url = False
mock_handler.document_url = "mailto:test@example.com"
mocker.patch("rasterize.navigate_to_path", return_value=mock_handler)
with capfd.disabled():
content, error_msg = extract_text_content(mock_browser, mock_tab, "mailto:test@example.com", 0, 30)
assert content is None
assert "Cannot rasterize" in error_msg
assert "mailto:" in error_msg
def test_extract_text_content_private_network(mocker, capfd):
"""
Given: A private network URL
When: Calling extract_text_content
Then: Should return None and error message
"""
from rasterize import extract_text_content
mock_browser = mocker.Mock()
mock_tab = mocker.Mock()
mock_tab.id = "test_tab_id"
mock_handler = mocker.Mock()
mock_handler.is_mailto = False
mock_handler.is_private_network_url = True
mock_handler.document_url = "http://192.168.1.1"
mocker.patch("rasterize.navigate_to_path", return_value=mock_handler)
with capfd.disabled():
content, error_msg = extract_text_content(mock_browser, mock_tab, "http://192.168.1.1", 0, 30)
assert content is None
assert "Cannot rasterize" in error_msg
assert "private network" in error_msg
def test_rasterize_extract_command_success(mocker):
"""
Given: Valid URLs for extraction
When: Calling rasterize_extract_command
Then: Should return CommandResults with extracted content
"""
from rasterize import rasterize_extract_command
mock_args = {"url": "https://example.com", "wait_time": "0", "max_page_load_time": "30"}
mocker.patch.object(demisto, "args", return_value=mock_args)
mocker.patch("rasterize.perform_rasterize", return_value=[("# Test Content", "https://example.com")])
mock_return_results = mocker.patch("rasterize.return_results")
rasterize_extract_command()
assert mock_return_results.called
results = mock_return_results.call_args[0][0]
assert len(results) == 1
assert results[0].outputs["URL"] == "https://example.com"
assert results[0].outputs["Content"] == "# Test Content"
def test_rasterize_extract_command_multiple_urls(mocker):
"""
Given: Multiple URLs for extraction
When: Calling rasterize_extract_command
Then: Should return CommandResults for each URL
"""
from rasterize import rasterize_extract_command
urls = ["https://example1.com", "https://example2.com"]
mock_args = {"url": urls, "wait_time": "0", "max_page_load_time": "30"}
mocker.patch.object(demisto, "args", return_value=mock_args)
mocker.patch(
"rasterize.perform_rasterize",
return_value=[("# Content 1", "https://example1.com"), ("# Content 2", "https://example2.com")],
)
mock_return_results = mocker.patch("rasterize.return_results")
rasterize_extract_command()
assert mock_return_results.called
results = mock_return_results.call_args[0][0]
assert len(results) == 2
assert results[0].outputs["URL"] == "https://example1.com"
assert results[1].outputs["URL"] == "https://example2.com"
def test_rasterize_extract_command_extraction_error(mocker):
"""
Given: A URL that fails extraction
When: Calling rasterize_extract_command
Then: Should return error CommandResults
"""
from rasterize import rasterize_extract_command
mock_args = {"url": "https://example.com", "wait_time": "0", "max_page_load_time": "30"}
mocker.patch.object(demisto, "args", return_value=mock_args)
mocker.patch("rasterize.perform_rasterize", return_value=[("Extraction Error: Test error", "https://example.com")])
mock_return_results = mocker.patch("rasterize.return_results")
rasterize_extract_command()
assert mock_return_results.called
results = mock_return_results.call_args[0][0]
assert len(results) == 1
assert results[0].entry_type == EntryType.ERROR
assert "Error extracting content" in results[0].readable_output
def test_rasterize_extract_command_string_error(mocker):
"""
Given: A URL that returns a string error
When: Calling rasterize_extract_command
Then: Should return error CommandResults
"""
from rasterize import rasterize_extract_command
mock_args = {"url": "https://example.com", "wait_time": "0", "max_page_load_time": "30"}
mocker.patch.object(demisto, "args", return_value=mock_args)
mocker.patch("rasterize.perform_rasterize", return_value=["Error: Connection failed"])
mock_return_results = mocker.patch("rasterize.return_results")
rasterize_extract_command()
assert mock_return_results.called
results = mock_return_results.call_args[0][0]
assert len(results) == 1
assert results[0].entry_type == EntryType.ERROR
assert "Error rasterizing" in results[0].readable_output
# region Memory Pressure Monitoring tests
def test_get_container_working_set_bytes_success(mocker):
"""
Given: cgroup v2 memory.current and memory.stat files with valid values.
When: Calling get_container_working_set_bytes.
Then: It returns memory.current minus inactive_file.
"""
from unittest.mock import mock_open
def fake_open(path, *args, **kwargs):
if path == "/sys/fs/cgroup/memory.current":
return mock_open(read_data="1000\n")()
if path == "/sys/fs/cgroup/memory.stat":
return mock_open(read_data="active_file 50\ninactive_file 300\nother 1\n")()
raise FileNotFoundError(path)
mocker.patch("builtins.open", side_effect=fake_open)
assert rasterize.get_container_working_set_bytes() == 700
def test_get_container_working_set_bytes_unreadable(mocker):
"""
Given: cgroup v2 files that cannot be read.
When: Calling get_container_working_set_bytes.
Then: It returns 0.
"""
mocker.patch("builtins.open", side_effect=FileNotFoundError)
assert rasterize.get_container_working_set_bytes() == 0
def test_get_container_available_memory_bytes_no_limit(mocker):
"""
Given: cgroup memory.max set to "max" (no hard limit).
When: Calling get_container_available_memory_bytes.
Then: It returns -1.
"""
from unittest.mock import mock_open
mocker.patch("builtins.open", mock_open(read_data="max\n"))
assert rasterize.get_container_available_memory_bytes() == -1
def test_get_container_available_memory_bytes_with_limit(mocker):
"""
Given: A hard memory limit and a known working set.
When: Calling get_container_available_memory_bytes.
Then: It returns max(0, mem_max - working_set).
"""
from unittest.mock import mock_open
mocker.patch("builtins.open", mock_open(read_data="1000\n"))
mocker.patch.object(rasterize, "get_container_working_set_bytes", return_value=300)
assert rasterize.get_container_available_memory_bytes() == 700
def test_get_container_available_memory_bytes_unreadable(mocker):
"""
Given: cgroup memory.max that cannot be read.
When: Calling get_container_available_memory_bytes.
Then: It returns 0.
"""
mocker.patch("builtins.open", side_effect=FileNotFoundError)
assert rasterize.get_container_available_memory_bytes() == 0
# endregion
# region CDP / freeze tests
def test_safe_call_cdp_with_args_none_tab():
"""
Given: A None tab.
When: Calling _safe_call_cdp_with_args.
Then: It is a no-op and does not raise.
"""
rasterize._safe_call_cdp_with_args(None, "Page.stopLoading", "tab_id", "path")
def test_safe_call_cdp_with_args_invokes_method(mocker):
"""
Given: A tab with a nested CDP method.
When: Calling _safe_call_cdp_with_args with kwargs.
Then: The resolved method is invoked with the forwarded kwargs.
"""
tab = mocker.MagicMock()
rasterize._safe_call_cdp_with_args(tab, "Network.emulateNetworkConditions", "tab_id", "path", offline=True)
tab.Network.emulateNetworkConditions.assert_called_once_with(offline=True)
def test_safe_call_cdp_with_args_swallows_exception(mocker):
"""
Given: A CDP method that raises.
When: Calling _safe_call_cdp_with_args.
Then: The exception is swallowed and not propagated.
"""
tab = mocker.MagicMock()
tab.Page.stopLoading.side_effect = Exception("disconnected")
# Should not raise
rasterize._safe_call_cdp_with_args(tab, "Page.stopLoading", "tab_id", "path")
def test_freeze_tab_for_screenshot_none_tab():
"""
Given: A None tab.
When: Calling _freeze_tab_for_screenshot.
Then: It is a no-op and does not raise.
"""
rasterize._freeze_tab_for_screenshot(None, "tab_id", "path")
def test_freeze_tab_for_screenshot_invokes_cdp_sequence(mocker):
"""
Given: A valid tab.
When: Calling _freeze_tab_for_screenshot.
Then: The full freeze CDP sequence is invoked via the safe helper.
"""
safe_call = mocker.patch.object(rasterize, "_safe_call_cdp_with_args")
mocker.patch.object(rasterize, "get_container_available_memory_bytes", return_value=1024 * 1024)
tab = mocker.MagicMock()
rasterize._freeze_tab_for_screenshot(tab, "tab_id", "path")
called_methods = [call.kwargs["method_path"] for call in safe_call.call_args_list]
for expected in [
"Network.enable",
"Network.emulateNetworkConditions",
"Fetch.enable",
"Page.stopLoading",
"Emulation.setScriptExecutionDisabled",
"Page.setWebLifecycleState",
"HeapProfiler.collectGarbage",
"Memory.forciblyPurgeJavaScriptMemory",
"Network.clearBrowserCache",
]:
assert expected in called_methods
# endregion
# region wait_for_page_load_with_memory_guard tests
def test_wait_for_page_load_no_cgroup_limit(mocker):
"""
Given: No cgroup memory limit (get_container_available_memory_bytes returns -1).
When: Calling wait_for_page_load_with_memory_guard.
Then: It performs a single blocking wait and returns True.
"""
mocker.patch.object(rasterize, "get_container_available_memory_bytes", return_value=-1)
event = threading.Event()
event.set()
freeze = mocker.patch.object(rasterize, "_freeze_tab_for_screenshot")
result = rasterize.wait_for_page_load_with_memory_guard(
tab_ready_event=event, navigation_timeout=1, tab_id="tab_id", path="path", tab=mocker.MagicMock()
)
assert result is True
# In the no-limit fast path the tab is not frozen.
freeze.assert_not_called()
def test_wait_for_page_load_normal_completion(mocker):
"""
Given: A cgroup limit exists and the page finishes loading (event already set).
When: Calling wait_for_page_load_with_memory_guard.
Then: It freezes the tab and returns True.
"""
mocker.patch.object(rasterize, "get_container_available_memory_bytes", return_value=10 * 1024 * 1024 * 1024)
freeze = mocker.patch.object(rasterize, "_freeze_tab_for_screenshot")
event = threading.Event()
event.set()
tab = mocker.MagicMock()
result = rasterize.wait_for_page_load_with_memory_guard(
tab_ready_event=event, navigation_timeout=5, tab_id="tab_id", path="path", tab=tab
)
assert result is True
freeze.assert_called_once_with(tab, "tab_id", "path")
def test_wait_for_page_load_memory_pressure(mocker):
"""
Given: A cgroup limit exists, the page does not finish loading, and available memory
drops below the tolerance.
When: Calling wait_for_page_load_with_memory_guard.
Then: It freezes the tab, sets the event and returns False (aborted early).
"""
# First call (initial -1 check) returns a positive value; subsequent samples are below tolerance.
mocker.patch.object(rasterize, "get_container_available_memory_bytes", return_value=10 * 1024 * 1024)
freeze = mocker.patch.object(rasterize, "_freeze_tab_for_screenshot")
mocker.patch.object(rasterize.time, "sleep") # avoid the 30s diagnostic sleep
event = threading.Event() # never set
tab = mocker.MagicMock()
result = rasterize.wait_for_page_load_with_memory_guard(
tab_ready_event=event,
navigation_timeout=5,
tolerance_bytes=650 * 1024 * 1024,
poll_interval=0.01,
tab_id="tab_id",
path="path",
tab=tab,
)
assert result is False
assert event.is_set()
freeze.assert_called_once_with(tab, "tab_id", "path")
def test_wait_for_page_load_timeout(mocker):
"""
Given: A cgroup limit exists, the page never finishes loading, but memory stays
above the tolerance until the navigation timeout elapses.
When: Calling wait_for_page_load_with_memory_guard.
Then: It stops loading via CDP and returns True (caller handles the timeout warning).
"""
mocker.patch.object(rasterize, "get_container_available_memory_bytes", return_value=10 * 1024 * 1024 * 1024)
safe_call = mocker.patch.object(rasterize, "_safe_call_cdp_with_args")
event = threading.Event() # never set
tab = mocker.MagicMock()
result = rasterize.wait_for_page_load_with_memory_guard(
tab_ready_event=event,
navigation_timeout=0,
tolerance_bytes=1,
poll_interval=0.01,
tab_id="tab_id",
path="path",
tab=tab,
)
assert result is True
stopped = [call.kwargs["method_path"] for call in safe_call.call_args_list]
assert "Page.stopLoading" in stopped
# endregion
def test_rasterize_extract_command_extraction_error_includes_guidance(mocker):
"""
Given: A URL that fails extraction with an "Extraction Error:" result
When: Calling rasterize_extract_command
Then: The error output includes actionable guidance (slow/anti-bot, retry with higher
max_page_load_time) so the caller/agent can act instead of retrying blindly
"""
from rasterize import rasterize_extract_command
mock_args = {"url": "https://example.com", "wait_time": "0", "max_page_load_time": "30"}
mocker.patch.object(demisto, "args", return_value=mock_args)
mocker.patch("rasterize.perform_rasterize", return_value=[("Extraction Error: timeout", "https://example.com")])
mock_return_results = mocker.patch("rasterize.return_results")
rasterize_extract_command()
output = mock_return_results.call_args[0][0][0].readable_output
assert "page-load timeout" in output
assert "anti-bot" in output
assert "max_page_load_time" in output
def test_rasterize_extract_command_passes_navigation_timeout_unchanged(mocker):
"""
Given: An explicit max_page_load_time
When: Calling rasterize_extract_command
Then: The navigation_timeout passed to perform_rasterize is the caller's value, unchanged
(the navigation budget is never shortened - slow pages still get their full time)
"""
from rasterize import rasterize_extract_command
mock_args = {"url": "https://example.com", "max_page_load_time": "250"}
mocker.patch.object(demisto, "args", return_value=mock_args)
mock_perform = mocker.patch("rasterize.perform_rasterize", return_value=[("# Content", "https://example.com")])
mocker.patch("rasterize.return_results")
rasterize_extract_command()
assert mock_perform.call_args.kwargs["navigation_timeout"] == 250
def test_extract_content_from_tab_uses_navigation_timeout(mocker):
"""
Given: A navigation_timeout value
When: Calling extract_content_from_tab
Then: The extraction JS timeout is the same navigation_timeout (original behavior - the total
per-URL budget is controlled via the action's max_page_load_time, not hardcoded here)
"""
from rasterize import extract_content_from_tab
mock_tab = mocker.Mock()
mock_tab.id = "test_tab_id"
mock_tab.Page.getFrameTree.return_value = {"frameTree": {"frame": {"url": "https://example.com"}}}
mock_tab.Runtime.evaluate.return_value = {"result": {"value": {"type": "html", "content": "# ok"}}}
extract_content_from_tab(mock_tab, 145)
assert mock_tab.Runtime.evaluate.call_args.kwargs["_timeout"] == 145