-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathsel-module.py
More file actions
116 lines (95 loc) · 4.47 KB
/
Copy pathsel-module.py
File metadata and controls
116 lines (95 loc) · 4.47 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
from os import path
from seleniumwire import webdriver
from selenium.webdriver.firefox.options import Options
import tools.turquoise_logger as turquoise_logger
import tools.mod_initializer as gui_enhancements
class SeleniumWireModule:
def __init__(self):
gui_enhancements.run_sel()
logg = turquoise_logger.Logger()
path_firefox_binary = 'tools/geckodriver.exe'
path_geckodriver_log = path.abspath('resources/geckodriver.log')
log = logg.logging()
localhost = '127.0.0.1'
initial_url = 'https://www.google.com'
options = Options()
options.add_argument('--headless')
options.set_preference('dom.webnotifications.enabled', False)
options.set_preference('dom.push.enabled', False)
options.set_preference('dom.webdriver.enabled', False)
options.set_preference('useAutomationExtension', False)
options.set_preference('privacy.trackingprotection.enabled', True)
options.set_preference('browser.cache.disk.enable', False)
options.set_preference('browser.cache.memory.enable', False)
options.set_preference('browser.cache.offline.enable', False)
options.set_preference('network.http.use-cache', False)
# USER AGENT OPT
# options.set_preference('intl.accept_languages', random.choice(localesList).lower())
# options.set_preference('general.useragent.override', random.choice(userAgentList))
profile_path = open('resources/profile_path', 'r').read()
driver = webdriver.Firefox(firefox_profile=profile_path,
options=options, executable_path=path_firefox_binary,
service_log_path=path_geckodriver_log)
driver.implicitly_wait(10)
log.debug(f'Webdriver is UP')
self.log = log
self.localhost = localhost
self.initial_url = initial_url
self.driver = driver
def healthcheck(self):
'''1 Returns an int if dead and None if alive
2 Throws a WebDriverException if dead'''
self.log.debug('Webdriver healthcheck going on')
try:
assert(self.driver.service.process.poll() == None)
self.driver.service.assert_process_still_running()
self.log.debug('The driver appears to be OK')
status = True
except Exception as e:
self.log.debug(f'The driver appears to be NOK - {e}')
status = False
return status
def connection_attempt(self, max_attempts_count=2):
'''Commits attempts_count connection attempts to the given initial_url'''
attempts_count = 1
initial_url=self.initial_url
while not attempts_count > max_attempts_count:
self.log.debug(f'{self.localhost} <-> {initial_url}')
self.driver.get(initial_url)
if len(self.driver.requests) > 0:
if self.driver.requests[0].response.status_code == 200:
self.log.debug(f'Connection reached | Attempts: {attempts_count}')
attempts_count = max_attempts_count
is_connected = True
else:
self.log.debug(f'Failed to connect | Attempts: {attempts_count}')
is_connected = False
attempts_count += 1
self.is_connected = is_connected
def requests_vars_get(self):
'''Outputs REQ (Request URL), STAT (Status Code) and CT (Content Type) within responses in requests'''
print('\n' + 'Responses summary')
if self.is_connected:
for request in self.driver.requests:
if request.response:
self.log.debug(f'REQ: {request.url}')
self.log.debug(f'STAT: {request.response.status_code}')
self.log.debug('CT:' + request.response.headers['content-type'] + '\n')
if not self.is_connected:
self.log.debug('The driver is not available')
def tearDown(self):
'''Graceful shutdown and status verification'''
self.driver.quit()
self.log.debug('Verifying webdriver shutdown')
status = self.healthcheck()
if status == False:
self.log.debug('Successful driver termination')
else:
self.log.error('Unsuccessful driver termination')
# TEST
wm = SeleniumWireModule()
wm_is_up = wm.healthcheck()
if wm_is_up:
is_connected = wm.connection_attempt()
wm.requests_vars_get()
wm.tearDown()