Skip to content
Projects
Groups
Snippets
Help
Loading...
Help
Support
Keyboard shortcuts
?
Submit feedback
Contribute to GitLab
Sign in / Register
Toggle navigation
S
slapos.toolbox
Project overview
Project overview
Details
Activity
Releases
Repository
Repository
Files
Commits
Branches
Tags
Contributors
Graph
Compare
Issues
0
Issues
0
List
Boards
Labels
Milestones
Merge Requests
5
Merge Requests
5
CI / CD
CI / CD
Pipelines
Jobs
Schedules
Analytics
Analytics
CI / CD
Repository
Value Stream
Wiki
Wiki
Members
Members
Collapse sidebar
Close sidebar
Activity
Graph
Create a new issue
Jobs
Commits
Issue Boards
Open sidebar
nexedi
slapos.toolbox
Commits
051f3471
Commit
051f3471
authored
Feb 07, 2019
by
Bryton Lacquement
🚪
Browse files
Options
Browse Files
Download
Plain Diff
Resolve conflicts introduced by
!50
parents
eb3a3bdd
9b4a5287
Changes
16
Hide whitespace changes
Inline
Side-by-side
Showing
16 changed files
with
920 additions
and
279 deletions
+920
-279
CHANGES.txt
CHANGES.txt
+5
-0
setup.py
setup.py
+2
-7
slapos/promise/check_error_on_apache_log/__init__.py
slapos/promise/check_error_on_apache_log/__init__.py
+0
-84
slapos/promise/check_re6st_optimal_status/__init__.py
slapos/promise/check_re6st_optimal_status/__init__.py
+0
-60
slapos/promise/is_icmp_packet_lost/__init__.py
slapos/promise/is_icmp_packet_lost/__init__.py
+0
-33
slapos/promise/plugin/check_error_on_http_log.py
slapos/promise/plugin/check_error_on_http_log.py
+84
-0
slapos/promise/plugin/check_icmp_packet_lost.py
slapos/promise/plugin/check_icmp_packet_lost.py
+48
-0
slapos/promise/plugin/check_re6st_optimal_status.py
slapos/promise/plugin/check_re6st_optimal_status.py
+66
-0
slapos/promise/plugin/check_url_available.py
slapos/promise/plugin/check_url_available.py
+62
-0
slapos/promise/plugin/trafficserver_cache_availability.py
slapos/promise/plugin/trafficserver_cache_availability.py
+37
-0
slapos/promise/plugin/validate_frontend_configuration.py
slapos/promise/plugin/validate_frontend_configuration.py
+36
-0
slapos/test/promise/plugin/test_check_error_on_http_log.py
slapos/test/promise/plugin/test_check_error_on_http_log.py
+209
-0
slapos/test/promise/plugin/test_check_icmp_packet_lost.py
slapos/test/promise/plugin/test_check_icmp_packet_lost.py
+159
-0
slapos/test/promise/plugin/test_check_re6st_optimal_status.py
...os/test/promise/plugin/test_check_re6st_optimal_status.py
+119
-0
slapos/test/promise/plugin/test_check_url_available.py
slapos/test/promise/plugin/test_check_url_available.py
+93
-0
slapos/test/promise/test_check_error_on_apache_log.py
slapos/test/promise/test_check_error_on_apache_log.py
+0
-95
No files found.
CHANGES.txt
View file @
051f3471
0.86 (2019-02-06)
=================
* promise: migrates old promises used in caddy-frontend SR to new format
0.85 (2019-01-09)
0.85 (2019-01-09)
=================
=================
...
...
setup.py
View file @
051f3471
...
@@ -2,7 +2,7 @@ from setuptools import setup, find_packages
...
@@ -2,7 +2,7 @@ from setuptools import setup, find_packages
import
glob
import
glob
import
os
import
os
version
=
'0.8
5
'
version
=
'0.8
6
'
name
=
'slapos.toolbox'
name
=
'slapos.toolbox'
long_description
=
open
(
"README.rst"
).
read
()
+
"
\
n
"
long_description
=
open
(
"README.rst"
).
read
()
+
"
\
n
"
...
@@ -57,17 +57,15 @@ setup(name=name,
...
@@ -57,17 +57,15 @@ setup(name=name,
'requests'
,
'requests'
,
'jsonschema'
,
'jsonschema'
,
'zc.buildout'
,
'zc.buildout'
,
'pycurl'
,
]
+
additional_install_requires
,
]
+
additional_install_requires
,
extras_require
=
{
extras_require
=
{
'lampconfigure'
:
[
"mysqlclient"
],
#needed for MySQL Database access
'lampconfigure'
:
[
"mysqlclient"
],
#needed for MySQL Database access
'zodbpack'
:
[
'ZODB3'
],
# needed to play with ZODB
'zodbpack'
:
[
'ZODB3'
],
# needed to play with ZODB
'flask_auth'
:
[
"Flask-Auth"
],
'flask_auth'
:
[
"Flask-Auth"
],
'networkbench'
:
[
'pycurl'
],
'check_web_page_http_cache_hit'
:
[
'pycurl'
],
# needed for check_web_page_http_cache_hit module
},
},
tests_require
=
[
tests_require
=
[
'mock'
,
'mock'
,
'pycurl'
],
],
zip_safe
=
False
,
# proxy depends on Flask, which has issues with
zip_safe
=
False
,
# proxy depends on Flask, which has issues with
# accessing templates
# accessing templates
...
@@ -78,10 +76,8 @@ setup(name=name,
...
@@ -78,10 +76,8 @@ setup(name=name,
'check-computer-memory = slapos.promise.check_computer_memory:main'
,
'check-computer-memory = slapos.promise.check_computer_memory:main'
,
'check-web-page-http-cache-hit = slapos.promise.check_web_page_http_cache_hit:main'
,
'check-web-page-http-cache-hit = slapos.promise.check_web_page_http_cache_hit:main'
,
'check-feed-as-promise = slapos.checkfeedaspromise:main'
,
'check-feed-as-promise = slapos.checkfeedaspromise:main'
,
'check-error-on-apache-log = slapos.promise.check_error_on_apache_log:main'
,
'check-apachedex-result = slapos.promise.check_apachedex_result:main'
,
'check-apachedex-result = slapos.promise.check_apachedex_result:main'
,
'check-slow-queries-digest-result = slapos.promise.check_slow_queries_digest_result:main'
,
'check-slow-queries-digest-result = slapos.promise.check_slow_queries_digest_result:main'
,
'check-re6st-optimal-status = slapos.promise.check_re6st_optimal_status:main'
,
'clouddestroy = slapos.cloudmgr.destroy:main'
,
'clouddestroy = slapos.cloudmgr.destroy:main'
,
'cloudgetprivatekey = slapos.cloudmgr.getprivatekey:main'
,
'cloudgetprivatekey = slapos.cloudmgr.getprivatekey:main'
,
'cloudgetpubliciplist = slapos.cloudmgr.getpubliciplist:main'
,
'cloudgetpubliciplist = slapos.cloudmgr.getpubliciplist:main'
,
...
@@ -94,7 +90,6 @@ setup(name=name,
...
@@ -94,7 +90,6 @@ setup(name=name,
'htpasswd = slapos.htpasswd:main'
,
'htpasswd = slapos.htpasswd:main'
,
'is-local-tcp-port-opened = slapos.promise.is_local_tcp_port_opened:main'
,
'is-local-tcp-port-opened = slapos.promise.is_local_tcp_port_opened:main'
,
'is-process-older-than-dependency-set = slapos.promise.is_process_older_than_dependency_set:main'
,
'is-process-older-than-dependency-set = slapos.promise.is_process_older_than_dependency_set:main'
,
'is-icmp-packet-lost = slapos.promise.is_icmp_packet_lost:main'
,
'killpidfromfile = slapos.systool:killpidfromfile'
,
# BBB
'killpidfromfile = slapos.systool:killpidfromfile'
,
# BBB
'monitor.bootstrap = slapos.monitor.monitor:main'
,
'monitor.bootstrap = slapos.monitor.monitor:main'
,
'monitor.collect = slapos.monitor.collect:main'
,
'monitor.collect = slapos.monitor.collect:main'
,
...
...
slapos/promise/check_error_on_apache_log/__init__.py
deleted
100644 → 0
View file @
eb3a3bdd
from
__future__
import
print_function
import
re
import
time
import
sys
import
gzip
import
argparse
import
os
r
=
re
.
compile
(
br"^(\
[[^
\]]+\
]) (
\[[^\
]]+
\]) (.*)$"
)
def
test
(
log_file
,
maximum_delay
):
error_amount
=
0
no_route_error
=
0
network_is_unreacheable
=
0
timeout
=
0
parsing_failure
=
0
if
not
os
.
path
.
exists
(
log_file
):
# file don't exist, nothing to check
return
"OK"
with
open
(
log_file
,
"rb"
)
as
f
:
f
.
seek
(
0
,
2
)
block_end_byte
=
f
.
tell
()
f
.
seek
(
-
min
(
block_end_byte
,
4096
),
1
)
data
=
f
.
read
()
for
line
in
reversed
(
data
.
splitlines
()):
m
=
r
.
match
(
line
)
if
m
is
None
:
continue
dt
,
level
,
msg
=
m
.
groups
()
try
:
try
:
t
=
time
.
strptime
(
dt
[
1
:
-
1
].
decode
(
'utf-8'
),
"%a %b %d %H:%M:%S %Y"
)
except
ValueError
:
# Fail to parser for the first time, try a different output.
t
=
time
.
strptime
(
dt
[
1
:
-
1
].
decode
(
'utf-8'
),
"%a %b %d %H:%M:%S.%f %Y"
)
except
ValueError
:
# Probably it fail to parse
if
parsing_failure
<
3
:
# Accept failure 2 times, as the line can be actually
# cut on the middle.
parsing_failure
+=
1
continue
raise
if
maximum_delay
and
(
time
.
time
()
-
time
.
mktime
(
t
))
>
maximum_delay
:
# no result in the latest hour
break
if
level
!=
b"[error]"
:
continue
# Classify the types of errors
if
b"(113)No route to host"
in
msg
:
no_route_error
+=
1
elif
b"(101)Network is unreachable"
in
msg
:
network_is_unreacheable
+=
1
elif
b"(110)Connection timed out"
in
msg
:
timeout
+=
1
error_amount
+=
1
if
error_amount
:
return
"ERROR=%s (NOTROUTE=%s, UNREACHEABLENET=%s, TIMEOUT=%s)"
%
(
error_amount
,
no_route_error
,
network_is_unreacheable
,
timeout
)
return
"OK"
def
main
():
parser
=
argparse
.
ArgumentParser
()
parser
.
add_argument
(
"-l"
,
"--log-file"
,
required
=
True
)
parser
.
add_argument
(
"-d"
,
"--maximum-delay"
,
default
=
0
)
args
=
parser
.
parse_args
()
log_file
=
args
.
log_file
result
=
test
(
args
.
log_file
,
args
.
maximum_delay
)
print
(
result
)
if
result
!=
"OK"
:
sys
.
exit
(
1
)
slapos/promise/check_re6st_optimal_status/__init__.py
deleted
100644 → 0
View file @
eb3a3bdd
from
__future__
import
print_function
import
argparse
import
re
import
time
import
sys
from
slapos.networkbench.ping
import
ping
,
ping6
def
test
(
ipv6
,
ipv4
,
count
):
result_ipv4
=
ping
(
ipv4
,
count
=
count
)
print
(
"%s host=%s code=%s, result=%s, packet_lost_ratio=%s msg=%s"
%
result_ipv4
)
result_ipv6
=
ping6
(
ipv6
,
count
=
count
)
print
(
"%s host=%s code=%s, result=%s, packet_lost_ratio=%s msg=%s"
%
result_ipv6
)
if
result_ipv4
[
3
]
==
"failed"
and
result_ipv6
[
3
]
!=
"failed"
:
# IPv4 is unreacheable
return
"OK"
if
result_ipv6
[
3
]
==
"failed"
:
# IPv6 is unreacheable
return
"FAILED"
latency4
=
float
(
result_ipv4
[
3
])
latency6
=
float
(
result_ipv6
[
3
])
# We can consider that at worst 1ms is added to
# ipv4 response, due the usage of openvpn.
acceptable_delay
=
1
# We can consider that we accept a certain increase
# on latency, if we are on a bit congested link.
# So 10% is reseonable enough.
acceptable_lost
=
0.10
# Increase latency with the value.
latency4
+=
acceptable_delay
+
latency4
*
acceptable_lost
if
latency4
<
latency6
:
print
(
"Fail %s (latency4) > %s (latence6)"
%
(
latency4
,
latency6
))
return
"FAIL"
# Compare if both has Same working rate
return
"OK"
def
main
():
parser
=
argparse
.
ArgumentParser
()
# promise ipv6 and ipv4 address to compare.
parser
.
add_argument
(
"-4"
,
"--ipv4"
,
required
=
1
)
parser
.
add_argument
(
"-6"
,
"--ipv6"
,
required
=
1
)
parser
.
add_argument
(
"-c"
,
"--count"
,
default
=
10
)
args
=
parser
.
parse_args
()
result
=
test
(
args
.
ipv6
,
args
.
ipv4
,
args
.
count
)
print
(
result
)
if
result
!=
"OK"
:
# re6st is not on an optimal state.
sys
.
exit
(
1
)
slapos/promise/is_icmp_packet_lost/__init__.py
deleted
100644 → 0
View file @
eb3a3bdd
from
__future__
import
print_function
import
argparse
import
re
import
time
import
sys
from
slapos.networkbench.ping
import
ping
,
ping6
def
test
(
address
,
ipv4
,
count
):
if
ipv4
:
return
ping
(
address
,
count
=
count
)
return
ping6
(
address
,
count
=
count
)
def
main
():
parser
=
argparse
.
ArgumentParser
()
# Address to ping to
parser
.
add_argument
(
"-a"
,
"--address"
,
required
=
True
)
# Force use ipv4 protocol
parser
.
add_argument
(
"-4"
,
"--ipv4"
,
action
=
"store_true"
)
parser
.
add_argument
(
"-c"
,
"--count"
,
metavar
=
"COUNT"
,
default
=
10
)
args
=
parser
.
parse_args
()
result
=
test
(
args
.
address
,
args
.
ipv4
,
args
.
count
)
print
(
"%s host=%s code=%s, result=%s, packet_lost_ratio=%s msg=%s"
%
result
)
if
result
[
4
]
!=
"0"
:
# Packet lost occurred
print
(
"FAIL"
)
sys
.
exit
(
1
)
print
(
"OK"
)
slapos/promise/plugin/check_error_on_http_log.py
0 → 100644
View file @
051f3471
from
zope.interface
import
implementer
from
slapos.grid.promise
import
interface
from
slapos.grid.promise.generic
import
GenericPromise
,
TestResult
import
re
import
time
import
os
@
implementer
(
interface
.
IPromise
)
class
RunPromise
(
GenericPromise
):
def
__init__
(
self
,
config
):
GenericPromise
.
__init__
(
self
,
config
)
# set periodicity to run the promise twice per day
self
.
custom_frequency
=
int
(
self
.
getConfig
(
'frequency'
,
720
))
self
.
setPeriodicity
(
self
.
custom_frequency
)
# Skip test check on this promise
self
.
setTestLess
()
def
sense
(
self
):
"""
Check if http log file contain errors
"""
log_file
=
self
.
getConfig
(
'log-file'
)
maximum_delay
=
int
(
self
.
getConfig
(
'maximum-delay'
,
0
))
if
not
log_file
:
raise
ValueError
(
"log file was not set in promise parameters."
)
regex
=
re
.
compile
(
br"^(\
[[^
\]]+\
]) (
\[[^\
]]+
\]) (.*)$"
)
error_amount
=
0
no_route_error
=
0
network_is_unreachable
=
0
timeout
=
0
parsing_failure
=
0
if
not
os
.
path
.
exists
(
log_file
):
# file don't exist, nothing to check
self
.
logger
.
info
(
"OK"
)
return
with
open
(
log_file
,
"rb"
)
as
f
:
f
.
seek
(
0
,
2
)
block_end_byte
=
f
.
tell
()
f
.
seek
(
-
min
(
block_end_byte
,
4096
),
1
)
data
=
f
.
read
()
for
line
in
reversed
(
data
.
splitlines
()):
m
=
regex
.
match
(
line
)
if
m
is
None
:
continue
dt
,
level
,
msg
=
m
.
groups
()
try
:
try
:
t
=
time
.
strptime
(
dt
[
1
:
-
1
].
decode
(
'utf-8'
),
"%a %b %d %H:%M:%S %Y"
)
except
ValueError
:
# Fail to parser for the first time, try a different output.
t
=
time
.
strptime
(
dt
[
1
:
-
1
].
decode
(
'utf-8'
),
"%a %b %d %H:%M:%S.%f %Y"
)
except
ValueError
:
# Probably it fail to parse
if
parsing_failure
<
3
:
# Accept failure 2 times, as the line can be actually
# cut on the middle.
parsing_failure
+=
1
continue
raise
if
maximum_delay
and
(
time
.
time
()
-
time
.
mktime
(
t
))
>
maximum_delay
:
# no result in the latest hour
break
if
level
!=
b"[error]"
:
continue
# Classify the types of errors
if
b"(113)No route to host"
in
msg
:
no_route_error
+=
1
elif
b"(101)Network is unreachable"
in
msg
:
network_is_unreachable
+=
1
elif
b"(110)Connection timed out"
in
msg
:
timeout
+=
1
error_amount
+=
1
if
error_amount
:
self
.
logger
.
error
(
"ERROR=%s (NOROUTE=%s, UNREACHABLENET=%s, TIMEOUT=%s)"
%
(
error_amount
,
no_route_error
,
network_is_unreachable
,
timeout
))
else
:
self
.
logger
.
info
(
"OK"
)
def
anomaly
(
self
):
# only check the result of the two latest sense call
return
self
.
_test
(
result_count
=
2
,
failure_amount
=
2
,
latest_minute
=
self
.
custom_frequency
*
3
)
slapos/promise/plugin/check_icmp_packet_lost.py
0 → 100644
View file @
051f3471
from
zope.interface
import
implementer
from
slapos.grid.promise
import
interface
from
slapos.grid.promise.generic
import
GenericPromise
,
TestResult
import
re
import
time
from
slapos.networkbench.ping
import
ping
,
ping6
@
implementer
(
interface
.
IPromise
)
class
RunPromise
(
GenericPromise
):
def
__init__
(
self
,
config
):
GenericPromise
.
__init__
(
self
,
config
)
# set periodicity to run the promise twice per day
self
.
custom_frequency
=
int
(
self
.
getConfig
(
'frequency'
,
720
))
self
.
setPeriodicity
(
self
.
custom_frequency
)
# Skip test check on this promise
self
.
setTestLess
()
def
sense
(
self
):
"""
Check if there ICMP packets lost on given address
"""
# Address to ping to
address
=
self
.
getConfig
(
'address'
)
if
not
address
:
raise
ValueError
(
"'address' was not set in promise parameters."
)
# Force use ipv4 protocol ?
ipv4
=
self
.
getConfig
(
'ipv4'
)
in
(
'True'
,
'true'
,
'1'
)
count
=
int
(
self
.
getConfig
(
'count'
,
10
))
threshold
=
int
(
self
.
getConfig
(
'threshold'
,
0
))
if
threshold
<
0
:
raise
ValueError
(
"'threshold' value should be greater than 0."
)
if
ipv4
:
result
=
ping
(
address
,
count
=
count
)
else
:
result
=
ping6
(
address
,
count
=
count
)
message
=
"%s host=%s code=%s, result=%s, packet_lost_ratio=%s msg=%s"
%
result
packet_lost_ratio
=
int
(
result
[
4
])
if
packet_lost_ratio
==
-
1
or
packet_lost_ratio
>
threshold
:
# Packet lost occurred
self
.
logger
.
error
(
message
)
else
:
self
.
logger
.
info
(
message
)
def
anomaly
(
self
):
# only check the result of the two latest sense call
return
self
.
_test
(
result_count
=
2
,
failure_amount
=
2
,
latest_minute
=
self
.
custom_frequency
*
3
)
slapos/promise/plugin/check_re6st_optimal_status.py
0 → 100644
View file @
051f3471
from
zope.interface
import
implementer
from
slapos.grid.promise
import
interface
from
slapos.grid.promise.generic
import
GenericPromise
,
TestResult
import
re
import
time
from
slapos.networkbench.ping
import
ping
,
ping6
@
implementer
(
interface
.
IPromise
)
class
RunPromise
(
GenericPromise
):
def
__init__
(
self
,
config
):
GenericPromise
.
__init__
(
self
,
config
)
# set periodicity to run the promise twice per day
self
.
custom_frequency
=
int
(
self
.
getConfig
(
'frequency'
,
720
))
self
.
setPeriodicity
(
self
.
custom_frequency
)
# Skip test check on this promise
self
.
setTestLess
()
def
sense
(
self
):
"""
Check re6st optimal status
"""
# promise ipv6 and ipv4 address to compare.
ipv4
=
self
.
getConfig
(
'ipv4'
)
ipv6
=
self
.
getConfig
(
'ipv6'
)
count
=
int
(
self
.
getConfig
(
'count'
,
10
))
if
not
ipv4
:
raise
ValueError
(
"'ipv4' was not set in promise parameters."
)
if
not
ipv6
:
raise
ValueError
(
"'ipv6' was not set in promise parameters."
)
result_ipv4
=
ping
(
ipv4
,
count
=
count
)
result_ipv6
=
ping6
(
ipv6
,
count
=
count
)
# push into to the log file
self
.
logger
.
info
(
"%s host=%s code=%s, result=%s, packet_lost_ratio=%s msg=%s"
%
result_ipv4
)
self
.
logger
.
info
(
"%s host=%s code=%s, result=%s, packet_lost_ratio=%s msg=%s"
%
result_ipv6
)
if
result_ipv4
[
3
]
==
"failed"
and
result_ipv6
[
3
]
!=
"failed"
:
# IPv4 is unreacheable
self
.
logger
.
info
(
"OK: IPv4 unreachable, IPv6 reachable"
)
return
if
result_ipv6
[
3
]
==
"failed"
:
# IPv6 is unreacheable
self
.
logger
.
error
(
"FAILED: IPv4 reachable, IPv6 unreachable"
)
return
latency4
=
float
(
result_ipv4
[
3
])
latency6
=
float
(
result_ipv6
[
3
])
# We can consider that at worst 1ms is added to
# ipv4 response, due the usage of openvpn.
acceptable_delay
=
int
(
self
.
getConfig
(
'acceptable-delay'
,
1
))
# We can consider that we accept a certain increase
# on latency, if we are on a bit congested link.
# So 10% is reseonable enough.
acceptable_lost
=
int
(
self
.
getConfig
(
'acceptable-lost'
,
0.10
))
# Increase latency with the value.
latency4
+=
acceptable_delay
+
latency4
*
acceptable_lost
if
latency4
<
latency6
:
self
.
logger
.
error
(
"FAIL %s (latency4) > %s (latence6)"
%
(
latency4
,
latency6
))
else
:
# Compare if both has Same working rate
self
.
logger
.
info
(
"OK: IPv4 reachable, IPv6 reachable"
)
def
anomaly
(
self
):
# only check the result of the two latest sense call
return
self
.
_test
(
result_count
=
2
,
failure_amount
=
2
,
latest_minute
=
self
.
custom_frequency
*
3
)
slapos/promise/plugin/check_url_available.py
0 → 100644
View file @
051f3471
from
zope.interface
import
implementer
from
slapos.grid.promise
import
interface
from
slapos.grid.promise.generic
import
GenericPromise
import
os
import
pycurl
@
implementer
(
interface
.
IPromise
)
class
RunPromise
(
GenericPromise
):
def
__init__
(
self
,
config
):
GenericPromise
.
__init__
(
self
,
config
)
# SR can set custom periodicity
self
.
setPeriodicity
(
float
(
self
.
getConfig
(
'frequency'
,
2
)))
def
sense
(
self
):
"""
Check if frontend URL is available
"""
url
=
self
.
getConfig
(
'url'
)
timeout
=
int
(
self
.
getConfig
(
'timeout'
,
20
))
expected_http_code
=
int
(
self
.
getConfig
(
'http_code'
,
'200'
))
curl
=
pycurl
.
Curl
()
curl
.
setopt
(
pycurl
.
URL
,
url
)
curl
.
setopt
(
pycurl
.
TIMEOUT
,
timeout
)
curl
.
setopt
(
pycurl
.
FOLLOWLOCATION
,
True
)
curl
.
setopt
(
pycurl
.
SSL_VERIFYPEER
,
False
)
curl
.
setopt
(
pycurl
.
SSL_VERIFYHOST
,
False
)
curl
.
setopt
(
pycurl
.
WRITEFUNCTION
,
lambda
x
:
None
)
ca_cert_file
=
self
.
getConfig
(
'ca-cert-file'
)
cert_file
=
self
.
getConfig
(
'cert-file'
)
key_file
=
self
.
getConfig
(
'key-file'
)
if
key_file
and
cert_file
and
ca_cert_file
:
# set certificate configuration
curl
.
setopt
(
curl
.
CAINFO
,
ca_cert_file
)
curl
.
setopt
(
curl
.
SSLCERT
,
cert_file
)
curl
.
setopt
(
curl
.
SSLKEY
,
key_file
)
try
:
curl
.
perform
()
except
pycurl
.
error
as
e
:
code
,
message
=
e
.
args
self
.
logger
.
error
(
"%s: %s"
%
(
code
,
message
))
return
http_code
=
curl
.
getinfo
(
pycurl
.
HTTP_CODE
)
check_secure
=
self
.
getConfig
(
'check-secure'
)
curl
.
close
()
if
http_code
==
0
:
self
.
logger
.
error
(
"%s is not available (server not reachable)."
%
url
)
elif
http_code
==
401
and
check_secure
==
"1"
:
self
.
logger
.
info
(
"%s is protected (returned %s)."
%
(
url
,
http_code
))
elif
http_code
!=
expected_http_code
:
self
.
logger
.
error
(
"%s is not available (returned %s, expected %s)."
%
(
url
,
http_code
,
expected_http_code
))
else
:
self
.
logger
.
info
(
"%s: URL is available"
%
http_code
)
def
anomaly
(
self
):
return
self
.
_test
(
result_count
=
3
,
failure_amount
=
3
)
slapos/promise/plugin/trafficserver_cache_availability.py
0 → 100644
View file @
051f3471
from
zope.interface
import
implementer
from
slapos.grid.promise
import
interface
from
slapos.grid.promise.generic
import
GenericPromise
try
:
import
subprocess32
as
subprocess
except
ImportError
:
import
subprocess
@
implementer
(
interface
.
IPromise
)
class
RunPromise
(
GenericPromise
):
def
__init__
(
self
,
config
):
GenericPromise
.
__init__
(
self
,
config
)
self
.
setPeriodicity
(
minute
=
int
(
self
.
getConfig
(
'frequency'
,
5
)))
def
sense
(
self
):
"""
Check trafficserver cache availability
"""
traffic_line
=
self
.
getConfig
(
'wrapper-path'
)
process
=
subprocess
.
Popen
(
[
traffic_line
,
'-r'
,
'proxy.node.cache.percent_free'
],
stdout
=
subprocess
.
PIPE
,
stderr
=
subprocess
.
STDOUT
,
)
result
=
process
.
communicate
()[
0
].
strip
()
if
process
.
returncode
==
0
:
self
.
logger
.
info
(
"OK"
)
else
:
self
.
logger
.
error
(
"Cache not available, availability: %s"
%
result
)
def
anomaly
(
self
):
"""
There is an anomaly if last 3 senses were bad.
"""
return
self
.
_anomaly
(
result_count
=
3
,
failure_amount
=
3
)
\ No newline at end of file
slapos/promise/plugin/validate_frontend_configuration.py
0 → 100644
View file @
051f3471
from
zope.interface
import
implementer
from
slapos.grid.promise
import
interface
from
slapos.grid.promise.generic
import
GenericPromise
try
:
import
subprocess32
as
subprocess
except
ImportError
:
import
subprocess
@
implementer
(
interface
.
IPromise
)
class
RunPromise
(
GenericPromise
):
def
__init__
(
self
,
config
):
GenericPromise
.
__init__
(
self
,
config
)
# check configuration every 5 minutes (only for anomaly)
self
.
setPeriodicity
(
minute
=
int
(
self
.
getConfig
(
'frequency'
,
5
)))
def
sense
(
self
):
"""
Run frontend validatation script
"""
validate_script
=
self
.
getConfig
(
'verification-script'
)
if
not
validate_script
:
raise
ValueError
(
"'verification-script' was not set in promise parameters."
)
process
=
subprocess
.
Popen
(
[
validate_script
],
stdout
=
subprocess
.
PIPE
,
stderr
=
subprocess
.
STDOUT
,
)
message
=
process
.
communicate
()[
0
]
if
process
.
returncode
==
0
:
self
.
logger
.
info
(
"OK"
)
else
:
self
.
logger
.
error
(
"%s"
%
message
)
def
anomaly
(
self
):
return
self
.
_anomaly
(
result_count
=
1
,
failure_amount
=
1
)
slapos/test/promise/plugin/test_check_error_on_http_log.py
0 → 100644
View file @
051f3471
##############################################################################
#
# Copyright (c) 2018 Vifib SARL and Contributors. All Rights Reserved.
#
# WARNING: This program as such is intended to be used by professional
# programmers who take the whole responsibility of assessing all potential
# consequences resulting from its eventual inadequacies and bugs
# End users who are looking for a ready-to-use solution with commercial
# guarantees and support are strongly adviced to contract a Free Software
# Service Company
#
# This program is Free Software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 3
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
#
##############################################################################
from
slapos.test.promise.plugin
import
TestPromisePluginMixin
import
os
import
time
from
slapos.test.promise
import
data
from
slapos.grid.promise
import
PromiseError
class
TestCheckErrorOnHttpLog
(
TestPromisePluginMixin
):
def
get_time
(
self
,
sec
):
return
time
.
strftime
(
"%a %b %d %H:%M:%S %Y"
,
time
.
localtime
(
time
.
time
()
-
sec
))
def
_update_logs
(
self
):
log_file_list
=
[
"apache_error_log"
,
"infoonly_error_log"
,
"timeout_error_log"
,
"unreachable_error_log"
]
i
=
600
for
log_file
in
log_file_list
:
new
=
""
old
=
""
with
open
(
self
.
base_path
+
"/"
+
log_file
)
as
f
:
for
line
in
f
:
new
+=
line
.
replace
(
"DATETIME"
,
self
.
get_time
(
i
))
old
+=
line
.
replace
(
"DATETIME"
,
self
.
get_time
(
i
+
3600
))
i
-=
1
with
open
(
self
.
base_path
+
"/SOFTINST-0_"
+
log_file
,
"w"
)
as
f
:
f
.
write
(
old
)
f
.
write
(
new
)
def
setUp
(
self
):
TestPromisePluginMixin
.
setUp
(
self
)
log_folder
=
os
.
path
.
join
(
self
.
partition_dir
,
'var/log'
)
os
.
makedirs
(
log_folder
)
self
.
base_path
=
"/"
.
join
(
data
.
__file__
.
split
(
"/"
)[:
-
1
])
self
.
_update_logs
()
self
.
promise_name
=
"check-error-on-apache-log.py"
self
.
promise_pyc
=
os
.
path
.
join
(
self
.
plugin_dir
,
self
.
promise_name
+
"c"
)
self
.
base_content
=
"""from slapos.promise.plugin.check_error_on_http_log import RunPromise
extra_config_dict = {
'log-file': '%(log_file)s',
'maximum-delay': '%(maximum_delay)s',
}
"""
def
tearDown
(
self
):
TestPromisePluginMixin
.
tearDown
(
self
)
def
test_no_error
(
self
):
self
.
base_path
=
"/"
.
join
(
data
.
__file__
.
split
(
"/"
)[:
-
1
])
content
=
self
.
base_content
%
{
'log_file'
:
self
.
base_path
+
"/SOFTINST-0_infoonly_error_log"
,
'maximum_delay'
:
0
}
self
.
writePromise
(
self
.
promise_name
,
content
)
self
.
configureLauncher
(
force
=
True
,
enable_anomaly
=
True
)
self
.
launcher
.
run
()
# run a second time to add more results
self
.
launcher
.
run
()
result
=
self
.
getPromiseResult
(
self
.
promise_name
)
self
.
assertEqual
(
result
[
'result'
][
'failed'
],
False
)
self
.
assertEqual
(
result
[
'result'
][
'message'
],
"OK"
)
content
=
self
.
base_content
%
{
'log_file'
:
self
.
base_path
+
"/SOFTINST-0_infoonly_error_log"
,
'maximum_delay'
:
3600
}
self
.
writePromise
(
self
.
promise_name
,
content
)
# remove previous pyc, so modified promise file will be used
if
os
.
path
.
exists
(
self
.
promise_pyc
):
os
.
unlink
(
self
.
promise_pyc
)
# Ignore periodicity of the promise
self
.
launcher
.
run
()
result
=
self
.
getPromiseResult
(
self
.
promise_name
)
self
.
assertEqual
(
result
[
'result'
][
'failed'
],
False
)
self
.
assertEqual
(
result
[
'result'
][
'message'
],
"OK"
)
def
test_error
(
self
):
self
.
base_path
=
"/"
.
join
(
data
.
__file__
.
split
(
"/"
)[:
-
1
])
content
=
self
.
base_content
%
{
'log_file'
:
self
.
base_path
+
"/SOFTINST-0_apache_error_log"
,
'maximum_delay'
:
0
}
self
.
writePromise
(
self
.
promise_name
,
content
)
self
.
configureLauncher
(
force
=
True
,
enable_anomaly
=
True
)
self
.
launcher
.
run
()
with
self
.
assertRaises
(
PromiseError
):
self
.
launcher
.
run
()
result
=
self
.
getPromiseResult
(
self
.
promise_name
)
self
.
assertEqual
(
result
[
'result'
][
'failed'
],
True
)
self
.
assertEqual
(
result
[
'result'
][
'message'
],
"ERROR=2 (NOROUTE=2, UNREACHABLENET=0, TIMEOUT=0)"
)
content
=
self
.
base_content
%
{
'log_file'
:
self
.
base_path
+
"/SOFTINST-0_apache_error_log"
,
'maximum_delay'
:
3600
}
self
.
writePromise
(
self
.
promise_name
,
content
)
# remove previous pyc, so modified promise file will be used
if
os
.
path
.
exists
(
self
.
promise_pyc
):
os
.
unlink
(
self
.
promise_pyc
)
# Ignore periodicity of the promise
with
self
.
assertRaises
(
PromiseError
):
self
.
launcher
.
run
()
result
=
self
.
getPromiseResult
(
self
.
promise_name
)
self
.
assertEqual
(
result
[
'result'
][
'failed'
],
True
)
self
.
assertEqual
(
result
[
'result'
][
'message'
],
"ERROR=1 (NOROUTE=1, UNREACHABLENET=0, TIMEOUT=0)"
)
def
test_error_timeout
(
self
):
self
.
base_path
=
"/"
.
join
(
data
.
__file__
.
split
(
"/"
)[:
-
1
])
content
=
self
.
base_content
%
{
'log_file'
:
self
.
base_path
+
"/SOFTINST-0_timeout_error_log"
,
'maximum_delay'
:
0
}
self
.
writePromise
(
self
.
promise_name
,
content
)
self
.
configureLauncher
(
force
=
True
,
enable_anomaly
=
True
)
self
.
launcher
.
run
()
with
self
.
assertRaises
(
PromiseError
):
self
.
launcher
.
run
()
result
=
self
.
getPromiseResult
(
self
.
promise_name
)
self
.
assertEqual
(
result
[
'result'
][
'failed'
],
True
)
self
.
assertEqual
(
result
[
'result'
][
'message'
],
"ERROR=4 (NOROUTE=0, UNREACHABLENET=0, TIMEOUT=4)"
)
content
=
self
.
base_content
%
{
'log_file'
:
self
.
base_path
+
"/SOFTINST-0_timeout_error_log"
,
'maximum_delay'
:
3600
}
self
.
writePromise
(
self
.
promise_name
,
content
)
# remove previous pyc, so modified promise file will be used
if
os
.
path
.
exists
(
self
.
promise_pyc
):
os
.
unlink
(
self
.
promise_pyc
)
# Ignore periodicity of the promise
with
self
.
assertRaises
(
PromiseError
):
self
.
launcher
.
run
()
result
=
self
.
getPromiseResult
(
self
.
promise_name
)
self
.
assertEqual
(
result
[
'result'
][
'failed'
],
True
)
self
.
assertEqual
(
result
[
'result'
][
'message'
],
"ERROR=2 (NOROUTE=0, UNREACHABLENET=0, TIMEOUT=2)"
)
def
test_error_unreacheabler
(
self
):
self
.
base_path
=
"/"
.
join
(
data
.
__file__
.
split
(
"/"
)[:
-
1
])
content
=
self
.
base_content
%
{
'log_file'
:
self
.
base_path
+
"/SOFTINST-0_unreachable_error_log"
,
'maximum_delay'
:
0
}
self
.
writePromise
(
self
.
promise_name
,
content
)
self
.
configureLauncher
(
force
=
True
,
enable_anomaly
=
True
)
self
.
launcher
.
run
()
with
self
.
assertRaises
(
PromiseError
):
self
.
launcher
.
run
()
result
=
self
.
getPromiseResult
(
self
.
promise_name
)
self
.
assertEqual
(
result
[
'result'
][
'failed'
],
True
)
self
.
assertEqual
(
result
[
'result'
][
'message'
],
"ERROR=11 (NOROUTE=0, UNREACHABLENET=11, TIMEOUT=0)"
)
content
=
self
.
base_content
%
{
'log_file'
:
self
.
base_path
+
"/SOFTINST-0_unreachable_error_log"
,
'maximum_delay'
:
3600
}
self
.
writePromise
(
self
.
promise_name
,
content
)
# remove previous pyc, so modified promise file will be used
if
os
.
path
.
exists
(
self
.
promise_pyc
):
os
.
unlink
(
self
.
promise_pyc
)
# Ignore periodicity of the promise
with
self
.
assertRaises
(
PromiseError
):
self
.
launcher
.
run
()
result
=
self
.
getPromiseResult
(
self
.
promise_name
)
self
.
assertEqual
(
result
[
'result'
][
'failed'
],
True
)
self
.
assertEqual
(
result
[
'result'
][
'message'
],
"ERROR=11 (NOROUTE=0, UNREACHABLENET=11, TIMEOUT=0)"
)
if
__name__
==
'__main__'
:
unittest
.
main
()
slapos/test/promise/plugin/test_check_icmp_packet_lost.py
0 → 100644
View file @
051f3471
##############################################################################
#
# Copyright (c) 2018 Vifib SARL and Contributors. All Rights Reserved.
#
# WARNING: This program as such is intended to be used by professional
# programmers who take the whole responsibility of assessing all potential
# consequences resulting from its eventual inadequacies and bugs
# End users who are looking for a ready-to-use solution with commercial
# guarantees and support are strongly adviced to contract a Free Software
# Service Company
#
# This program is Free Software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 3
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
#
##############################################################################
from
slapos.test.promise.plugin
import
TestPromisePluginMixin
from
slapos.grid.promise
import
PromiseError
import
os
import
os.path
import
socket
class
TestCheckICMPPacketLost
(
TestPromisePluginMixin
):
def
setUp
(
self
):
TestPromisePluginMixin
.
setUp
(
self
)
log_folder
=
os
.
path
.
join
(
self
.
partition_dir
,
'var/log'
)
os
.
makedirs
(
log_folder
)
self
.
promise_name
=
"check-icmp-packet-lost.py"
self
.
base_content
=
"""from slapos.promise.plugin.check_icmp_packet_lost import RunPromise
extra_config_dict = {
'ipv4': '%(ipv4)s',
'count': '%(count)s',
'address': '%(address)s',
'threshold': '%(threshold)s'
}
"""
def
tearDown
(
self
):
TestPromisePluginMixin
.
tearDown
(
self
)
def
test_localhost
(
self
):
content
=
self
.
base_content
%
{
'address'
:
"localhost"
,
'count'
:
5
,
"ipv4"
:
True
,
"threshold"
:
0
}
self
.
writePromise
(
self
.
promise_name
,
content
)
self
.
configureLauncher
(
force
=
True
,
timeout
=
20
,
enable_anomaly
=
True
)
self
.
launcher
.
run
()
# run a second time to add more results
self
.
launcher
.
run
()
result
=
self
.
getPromiseResult
(
self
.
promise_name
)
self
.
assertEqual
(
result
[
'result'
][
'failed'
],
False
)
self
.
assertTrue
(
'packet_lost_ratio=0'
in
result
[
'result'
][
'message'
])
def
test_error
(
self
):
content
=
self
.
base_content
%
{
'address'
:
"couscous"
,
'count'
:
5
,
"ipv4"
:
True
,
"threshold"
:
0
}
self
.
writePromise
(
self
.
promise_name
,
content
)
self
.
configureLauncher
(
force
=
True
,
timeout
=
20
,
enable_anomaly
=
True
)
self
.
launcher
.
run
()
with
self
.
assertRaises
(
PromiseError
):
self
.
launcher
.
run
()
result
=
self
.
getPromiseResult
(
self
.
promise_name
)
self
.
assertEqual
(
result
[
'result'
][
'failed'
],
True
)
self
.
assertTrue
(
'packet_lost_ratio=-1'
in
result
[
'result'
][
'message'
])
def
test_localhost6_with_ping6
(
self
):
content
=
self
.
base_content
%
{
'address'
:
"::1"
,
'count'
:
5
,
"ipv4"
:
False
,
"threshold"
:
0
}
self
.
writePromise
(
self
.
promise_name
,
content
)
self
.
configureLauncher
(
force
=
True
,
timeout
=
20
,
enable_anomaly
=
True
)
self
.
launcher
.
run
()
# run a second time to add more results
self
.
launcher
.
run
()
result
=
self
.
getPromiseResult
(
self
.
promise_name
)
self
.
assertEqual
(
result
[
'result'
][
'failed'
],
False
)
self
.
assertTrue
(
'packet_lost_ratio=0'
in
result
[
'result'
][
'message'
])
def
test_localhost6_with_ping4
(
self
):
content
=
self
.
base_content
%
{
'address'
:
"::1"
,
'count'
:
5
,
"ipv4"
:
True
,
"threshold"
:
0
}
self
.
writePromise
(
self
.
promise_name
,
content
)
self
.
configureLauncher
(
force
=
True
,
timeout
=
20
,
enable_anomaly
=
True
)
self
.
launcher
.
run
()
# run a second time to add more results
self
.
launcher
.
run
()
result
=
self
.
getPromiseResult
(
self
.
promise_name
)
self
.
assertEqual
(
result
[
'result'
][
'failed'
],
False
)
self
.
assertTrue
(
'packet_lost_ratio=0'
in
result
[
'result'
][
'message'
])
def
test_error6
(
self
):
content
=
self
.
base_content
%
{
'address'
:
"couscous"
,
'count'
:
5
,
"ipv4"
:
False
,
"threshold"
:
0
}
self
.
writePromise
(
self
.
promise_name
,
content
)
self
.
configureLauncher
(
force
=
True
,
timeout
=
20
,
enable_anomaly
=
True
)
self
.
launcher
.
run
()
with
self
.
assertRaises
(
PromiseError
):
self
.
launcher
.
run
()
result
=
self
.
getPromiseResult
(
self
.
promise_name
)
self
.
assertEqual
(
result
[
'result'
][
'failed'
],
True
)
self
.
assertTrue
(
'packet_lost_ratio=-1'
in
result
[
'result'
][
'message'
])
def
test_packet_lost_less_than_threshold
(
self
):
content
=
self
.
base_content
%
{
'address'
:
"10.2.3.4"
,
'count'
:
5
,
"ipv4"
:
True
,
"threshold"
:
110
}
self
.
writePromise
(
self
.
promise_name
,
content
)
self
.
configureLauncher
(
force
=
True
,
timeout
=
20
,
enable_anomaly
=
True
)
self
.
launcher
.
run
()
# run a second time to add more results
self
.
launcher
.
run
()
result
=
self
.
getPromiseResult
(
self
.
promise_name
)
self
.
assertEqual
(
result
[
'result'
][
'failed'
],
False
)
self
.
assertTrue
(
'packet_lost_ratio=100'
in
result
[
'result'
][
'message'
])
if
__name__
==
'__main__'
:
unittest
.
main
()
slapos/test/promise/test_check_re6st_optimal_status.py
→
slapos/test/promise/
plugin/
test_check_re6st_optimal_status.py
View file @
051f3471
##############################################################################
##############################################################################
#
#
# Copyright (c) 201
7
Vifib SARL and Contributors. All Rights Reserved.
# Copyright (c) 201
8
Vifib SARL and Contributors. All Rights Reserved.
#
#
# WARNING: This program as such is intended to be used by professional
# WARNING: This program as such is intended to be used by professional
# programmers who take the whole responsibility of assessing all potential
# programmers who take the whole responsibility of assessing all potential
...
@@ -25,29 +25,95 @@
...
@@ -25,29 +25,95 @@
#
#
##############################################################################
##############################################################################
import
unittest
from
slapos.test.promise.plugin
import
TestPromisePluginMixin
import
os.path
from
slapos.grid.promise
import
PromiseError
import
socket
import
os
import
time
from
slapos.promise.check_re6st_optimal_status
import
test
class
TestCheckRe6stOptimalStatus
(
unittest
.
TestCase
):
class
TestCheckRe6stOptimalStatus
(
TestPromisePluginMixin
):
def
setUp
(
self
):
TestPromisePluginMixin
.
setUp
(
self
)
self
.
promise_name
=
"check-icmp-packet-lost.py"
self
.
base_content
=
"""from slapos.promise.plugin.check_re6st_optimal_status import RunPromise
extra_config_dict = {
'ipv4': '%(ipv4)s',
'count': '%(count)s',
'ipv6': '%(ipv6)s',
}
"""
def
tearDown
(
self
):
TestPromisePluginMixin
.
tearDown
(
self
)
def
test_ipv6_is_faster
(
self
):
def
test_ipv6_is_faster
(
self
):
result
=
test
(
'::1'
,
'8.8.8.8'
,
5
)
content
=
self
.
base_content
%
{
self
.
assertEqual
(
result
,
'OK'
)
'ipv4'
:
"8.8.8.8"
,
'ipv6'
:
'::1'
,
'count'
:
5
,
}
self
.
writePromise
(
self
.
promise_name
,
content
)
self
.
configureLauncher
(
force
=
True
,
timeout
=
20
,
enable_anomaly
=
True
)
self
.
launcher
.
run
()
# run a second time to add more results
self
.
launcher
.
run
()
result
=
self
.
getPromiseResult
(
self
.
promise_name
)
last_message
=
result
[
'result'
][
'message'
].
split
(
'
\
n
'
)[
-
1
]
self
.
assertEqual
(
result
[
'result'
][
'failed'
],
False
)
self
.
assertEqual
(
last_message
,
"OK: IPv4 reachable, IPv6 reachable"
)
def
test_ipv4_is_faster
(
self
):
def
test_ipv4_is_faster
(
self
):
result
=
test
(
'2001:67c:1254::1'
,
'127.0.0.1'
,
5
)
content
=
self
.
base_content
%
{
self
.
assertEqual
(
result
,
'FAIL'
)
'ipv4'
:
"127.0.0.1"
,
'ipv6'
:
'2001:67c:1254::1'
,
'count'
:
5
,
}
self
.
writePromise
(
self
.
promise_name
,
content
)
self
.
configureLauncher
(
force
=
True
,
timeout
=
20
,
enable_anomaly
=
True
)
self
.
launcher
.
run
()
with
self
.
assertRaises
(
PromiseError
):
self
.
launcher
.
run
()
result
=
self
.
getPromiseResult
(
self
.
promise_name
)
last_message
=
result
[
'result'
][
'message'
].
split
(
'
\
n
'
)[
-
1
]
self
.
assertEqual
(
result
[
'result'
][
'failed'
],
True
)
self
.
assertTrue
(
last_message
.
startswith
(
'FAIL'
))
def
test_ipv4_unreacheable_and_ipv6_ok
(
self
):
def
test_ipv4_unreacheable_and_ipv6_ok
(
self
):
result
=
test
(
'::1'
,
'couscous'
,
5
)
content
=
self
.
base_content
%
{
self
.
assertEqual
(
result
,
'OK'
)
'ipv4'
:
"couscous"
,
'ipv6'
:
'::1'
,
'count'
:
5
,
}
self
.
writePromise
(
self
.
promise_name
,
content
)
self
.
configureLauncher
(
force
=
True
,
timeout
=
20
,
enable_anomaly
=
True
)
self
.
launcher
.
run
()
# run a second time to add more results
self
.
launcher
.
run
()
result
=
self
.
getPromiseResult
(
self
.
promise_name
)
last_message
=
result
[
'result'
][
'message'
].
split
(
'
\
n
'
)[
-
1
]
self
.
assertEqual
(
result
[
'result'
][
'failed'
],
False
)
self
.
assertEqual
(
last_message
,
"OK: IPv4 unreachable, IPv6 reachable"
)
def
test_ipv6_fail
(
self
):
def
test_ipv6_fail
(
self
):
result
=
test
(
'couscous'
,
'127.0.0.1'
,
5
)
content
=
self
.
base_content
%
{
self
.
assertEqual
(
result
,
'FAILED'
)
'ipv4'
:
"127.0.0.1"
,
'ipv6'
:
'couscous'
,
'count'
:
5
,
}
self
.
writePromise
(
self
.
promise_name
,
content
)
self
.
configureLauncher
(
force
=
True
,
timeout
=
20
,
enable_anomaly
=
True
)
self
.
launcher
.
run
()
with
self
.
assertRaises
(
PromiseError
):
self
.
launcher
.
run
()
result
=
self
.
getPromiseResult
(
self
.
promise_name
)
last_message
=
result
[
'result'
][
'message'
].
split
(
'
\
n
'
)[
-
1
]
self
.
assertEqual
(
result
[
'result'
][
'failed'
],
True
)
self
.
assertEqual
(
last_message
,
"FAILED: IPv4 reachable, IPv6 unreachable"
)
if
__name__
==
'__main__'
:
if
__name__
==
'__main__'
:
unittest
.
main
()
unittest
.
main
()
slapos/test/promise/
test_is_icmp_packet_lost
.py
→
slapos/test/promise/
plugin/test_check_url_available
.py
View file @
051f3471
##############################################################################
##############################################################################
#
#
# Copyright (c) 201
7
Vifib SARL and Contributors. All Rights Reserved.
# Copyright (c) 201
8
Vifib SARL and Contributors. All Rights Reserved.
#
#
# WARNING: This program as such is intended to be used by professional
# WARNING: This program as such is intended to be used by professional
# programmers who take the whole responsibility of assessing all potential
# programmers who take the whole responsibility of assessing all potential
...
@@ -25,34 +25,69 @@
...
@@ -25,34 +25,69 @@
#
#
##############################################################################
##############################################################################
import
unittest
from
slapos.test.promise.plugin
import
TestPromisePluginMixin
import
os.path
from
slapos.grid.promise
import
PromiseError
import
socket
import
os
import
time
from
slapos.promise.is_icmp_packet_lost
import
test
class
TestCheckUrlAvailable
(
TestPromisePluginMixin
):
class
TestIsICMPPacketLost
(
unittest
.
TestCase
):
def
setUp
(
self
):
TestPromisePluginMixin
.
setUp
(
self
)
def
test_localhost
(
self
):
self
.
promise_name
=
"check-url-available.py"
result
=
test
(
"localhost"
,
True
,
5
)
self
.
assertEqual
(
result
[
4
],
'0'
)
self
.
base_content
=
"""from slapos.promise.plugin.check_url_available import RunPromise
def
test_error
(
self
):
extra_config_dict = {
result
=
test
(
"couscous"
,
True
,
5
)
'url': '%(url)s',
self
.
assertEqual
(
result
[
4
],
-
1
)
'timeout': %(timeout)s,
'check-secure': %(check_secure)s,
def
test_localhost6_with_ping6
(
self
):
}
result
=
test
(
"::1"
,
False
,
5
)
"""
self
.
assertEqual
(
result
[
4
],
'0'
)
def
tearDown
(
self
):
def
test_localhost6_with_ping4
(
self
):
TestPromisePluginMixin
.
tearDown
(
self
)
result
=
test
(
"::1"
,
True
,
5
)
self
.
assertEqual
(
result
[
4
],
'0'
)
def
test_check_url_bad
(
self
):
content
=
self
.
base_content
%
{
def
test_error6
(
self
):
'url'
:
'https://'
,
result
=
test
(
"couscous"
,
False
,
5
)
'timeout'
:
10
,
self
.
assertEqual
(
result
[
4
],
-
1
)
'check_secure'
:
0
}
self
.
writePromise
(
self
.
promise_name
,
content
)
self
.
configureLauncher
()
with
self
.
assertRaises
(
PromiseError
):
self
.
launcher
.
run
()
result
=
self
.
getPromiseResult
(
self
.
promise_name
)
self
.
assertEqual
(
result
[
'result'
][
'failed'
],
True
)
self
.
assertEqual
(
result
[
'result'
][
'message'
],
"3: Bad URL"
)
def
test_check_url_malformed
(
self
):
content
=
self
.
base_content
%
{
'url'
:
''
,
'timeout'
:
10
,
'check_secure'
:
0
}
self
.
writePromise
(
self
.
promise_name
,
content
)
self
.
configureLauncher
()
with
self
.
assertRaises
(
PromiseError
):
self
.
launcher
.
run
()
result
=
self
.
getPromiseResult
(
self
.
promise_name
)
self
.
assertEqual
(
result
[
'result'
][
'failed'
],
True
)
self
.
assertEqual
(
result
[
'result'
][
'message'
],
"3: <url> malformed"
)
def
test_check_url_site_off
(
self
):
content
=
content
=
self
.
base_content
%
{
'url'
:
'https://localhost:56789/site'
,
'timeout'
:
10
,
'check_secure'
:
0
}
self
.
writePromise
(
self
.
promise_name
,
content
)
self
.
configureLauncher
()
with
self
.
assertRaises
(
PromiseError
):
self
.
launcher
.
run
()
result
=
self
.
getPromiseResult
(
self
.
promise_name
)
self
.
assertEqual
(
result
[
'result'
][
'failed'
],
True
)
self
.
assertEqual
(
result
[
'result'
][
'message'
],
"7: Failed to connect to localhost port 56789: Connection refused"
)
if
__name__
==
'__main__'
:
if
__name__
==
'__main__'
:
unittest
.
main
()
unittest
.
main
()
slapos/test/promise/test_check_error_on_apache_log.py
deleted
100644 → 0
View file @
eb3a3bdd
##############################################################################
#
# Copyright (c) 2017 Vifib SARL and Contributors. All Rights Reserved.
#
# WARNING: This program as such is intended to be used by professional
# programmers who take the whole responsibility of assessing all potential
# consequences resulting from its eventual inadequacies and bugs
# End users who are looking for a ready-to-use solution with commercial
# guarantees and support are strongly adviced to contract a Free Software
# Service Company
#
# This program is Free Software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 3
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
#
##############################################################################
import
unittest
import
os.path
import
socket
import
time
from
slapos.promise.check_error_on_apache_log
import
test
from
slapos.test.promise
import
data
class
TestCheckErrorOnApacheLog
(
unittest
.
TestCase
):
def
get_time
(
self
,
sec
):
return
time
.
strftime
(
"%a %b %d %H:%M:%S %Y"
,
time
.
localtime
(
time
.
time
()
-
sec
))
def
_update_logs
(
self
):
log_file_list
=
[
"apache_error_log"
,
"infoonly_error_log"
,
"timeout_error_log"
,
"unreachable_error_log"
]
i
=
600
for
log_file
in
log_file_list
:
new
=
""
old
=
""
with
open
(
self
.
base_path
+
"/"
+
log_file
)
as
f
:
for
line
in
f
:
new
+=
line
.
replace
(
"DATETIME"
,
self
.
get_time
(
i
))
old
+=
line
.
replace
(
"DATETIME"
,
self
.
get_time
(
i
+
3600
))
i
-=
1
with
open
(
self
.
base_path
+
"/SOFTINST-0_"
+
log_file
,
"w"
)
as
f
:
f
.
write
(
old
)
f
.
write
(
new
)
def
setUp
(
self
):
self
.
base_path
=
"/"
.
join
(
data
.
__file__
.
split
(
"/"
)[:
-
1
])
self
.
_update_logs
()
def
test_no_error
(
self
):
self
.
assertEqual
(
"OK"
,
test
(
self
.
base_path
+
"/SOFTINST-0_infoonly_error_log"
,
0
))
self
.
assertEqual
(
"OK"
,
test
(
self
.
base_path
+
"/SOFTINST-0_infoonly_error_log"
,
3600
))
def
test_error
(
self
):
self
.
assertEqual
(
"ERROR=2 (NOTROUTE=2, UNREACHEABLENET=0, TIMEOUT=0)"
,
test
(
self
.
base_path
+
"/SOFTINST-0_apache_error_log"
,
0
))
self
.
assertEqual
(
"ERROR=1 (NOTROUTE=1, UNREACHEABLENET=0, TIMEOUT=0)"
,
test
(
self
.
base_path
+
"/SOFTINST-0_apache_error_log"
,
3600
))
def
test_error_timeout
(
self
):
self
.
assertEqual
(
"ERROR=4 (NOTROUTE=0, UNREACHEABLENET=0, TIMEOUT=4)"
,
test
(
self
.
base_path
+
"/SOFTINST-0_timeout_error_log"
,
0
))
self
.
assertEqual
(
"ERROR=2 (NOTROUTE=0, UNREACHEABLENET=0, TIMEOUT=2)"
,
test
(
self
.
base_path
+
"/SOFTINST-0_timeout_error_log"
,
3600
))
def
test_error_unreacheabler
(
self
):
self
.
assertEqual
(
"ERROR=11 (NOTROUTE=0, UNREACHEABLENET=11, TIMEOUT=0)"
,
test
(
self
.
base_path
+
"/SOFTINST-0_unreachable_error_log"
,
0
))
self
.
assertEqual
(
"ERROR=11 (NOTROUTE=0, UNREACHEABLENET=11, TIMEOUT=0)"
,
test
(
self
.
base_path
+
"/SOFTINST-0_unreachable_error_log"
,
3600
))
if
__name__
==
'__main__'
:
unittest
.
main
()
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment