availcheck.checker

  1# -*- coding: utf-8 -*-
  2# Copyright (C) 2024 TU Dresden
  3# ralf.klammer@tu-dresden.de
  4
  5import logging
  6
  7from json import dumps
  8from time import time
  9
 10from . import (
 11    BROWSER_DIR,
 12    O4A_ENTITIES_COUNT,
 13)
 14from .selenium import SeleniumAgent
 15from .configs import CheckerConfigs
 16
 17log = logging.getLogger(__name__)
 18
 19
 20def resolve_url(func):
 21    """
 22    Decorator to resolve the URL before running any check.
 23    """
 24
 25    def wrap(
 26        self,
 27        subdomain=None,
 28        path=None,
 29        ignore_shadow_root=False,
 30        *args,
 31        **kwargs,
 32    ):
 33        current_url = self.selenium.get_url()
 34        self.url = self.create_var_url(subdomain=subdomain, path=path)
 35        if current_url != self.url:
 36            try:
 37                self.selenium.get(self.url)
 38                if (
 39                    getattr(self, "shadow_root", None)
 40                    and ignore_shadow_root is not True
 41                ):
 42                    self.shadow_host = self.selenium.get_shadow_root(
 43                        self.shadow_root
 44                    )
 45            except Exception as e:
 46                log.error(f"Unable to open URL: {self.url}")
 47                log.error(e)
 48                return False
 49
 50            return func(self, *args, **kwargs)
 51        else:
 52            return func(self, *args, **kwargs)
 53
 54    return wrap
 55
 56
 57class CheckerBase(object):
 58    """
 59    Base class for all the availability checker
 60    """
 61
 62    def __init__(self, url=None, instance="prod", path_of_screenshots=None):
 63        self.instance = instance
 64        self.selenium = SeleniumAgent(
 65            path_of_screenshots=path_of_screenshots,
 66            implicit_wait=10,
 67            browser_dir=(
 68                f"{BROWSER_DIR}_{self.__class__.__name__}_{self.instance}"
 69                if BROWSER_DIR
 70                else None
 71            ),
 72        )
 73
 74        self.url = url
 75        self._checker_config = None
 76
 77    @property
 78    def checker_config(self):
 79        """
 80        Get the checker configs for the given instance.
 81        """
 82        return self._checker_config
 83
 84    @checker_config.setter
 85    def checker_config(self, value):
 86        """
 87        Set the checker configs for the given instance.
 88        """
 89        self._checker_config = value
 90
 91    @property
 92    def checks(self):
 93        """
 94        Get the relevant checks for the given instance from the checker configs.
 95        """
 96        if self.checker_config is not None:
 97            return self.checker_config.get_checks()
 98
 99    def get_base_url(self):
100        """
101        Get the base URL for the given instance from the checker configs.
102        """
103        if self.checker_config is not None:
104            return self.checker_config.get_url(self.instance)
105
106    def run_check(self, check, checkmk_format=True):
107        """
108        Run a single check for the given instance.
109        """
110        # check if the check is supported
111        if check not in self.checks:
112            raise ValueError(f"Requested check '{check}' does not exist.")
113        elif self.instance not in self.checks[check]["instances"]:
114            raise ValueError(
115                f"Requested check '{check}' is not supported for instance "
116                f"'{self.instance}'."
117            )
118        elif checkmk_format:
119            result = getattr(self, check)()
120            start_time = time()
121            result = getattr(self, check)()
122            execution_time = round(time() - start_time, 4)
123            status_code = (
124                0 if result else self.checks[check]["status_on_error"]
125            )
126            return "%i 'AvailabilityCheck | %s-%s | %s' duration=%s %s" % (
127                status_code,
128                self.checker_config.id,
129                self.instance,
130                check,
131                execution_time,
132                f'{self.checks[check]["description"]} (URL: {self.url} )',
133            )
134        else:
135            return getattr(self, check)()
136
137    def run_checks(self):
138        """
139        Run all checks defined in `self.checks` for the given instance.
140        """
141        for check in self.checks:
142            if self.instance not in self.checks[check]["instances"]:
143                continue
144            yield self.run_check(check)
145
146    def save_checks(self, path):
147        """
148        Save the results of the checks to a file.
149        """
150        results = self.run_checks()
151        filename = f"{self.__class__.__name__}_{self.instance}.json"
152        with open(f"{path.rstrip('/')}/{filename}", "w") as f:
153            f.write(dumps(results))
154        return filename
155
156    def create_var_url(
157        self,
158        subdomain=None,
159        path=None,
160    ):
161        """
162        Cutting the url and setting subdomains based
163        if production or testing instance set.
164        """
165
166        base_url = self.get_base_url()
167
168        if base_url is None:
169            raise ValueError(
170                f"self.Instance '{self.instance}' is not supported. "
171                "Please provide a valid URL."
172            )
173
174        url = base_url
175        if subdomain is not None:
176            url = f"https://{subdomain}.{base_url.split('://')[1]}"
177        if path is not None:
178            url = f"{url}/{path.lstrip('/')}"
179
180        return url
181
182    def quit(self):
183        self.selenium.quit()
184
185
186class CheckerWebApps(CheckerBase):
187    """
188    Checker for the Webapps Service
189    """
190
191    checker_config = CheckerConfigs("webapps")
192
193    @resolve_url
194    def landing_page_webapps(self):
195        """
196        Checks if the landing Page for Webapps is available
197        """
198
199        result = self.selenium.find_by_css(
200            css="img[alt='Logo des NFDI Konsortiums NFDI4Earth']"
201        )
202
203        return bool(result)
204
205    @resolve_url
206    def _visual_search_engine(self):
207        """
208        Checks if the Graph Based Visual Search Engine is available
209        """
210        result = self.selenium.wait_until(
211            css=".MuiDataGrid-cell--withRenderer", want_bool=True
212        )
213
214        result = self.selenium.amount_of_elements(
215            css=".MuiDataGrid-cell--withRenderer", minimum=2
216        )
217
218        return bool(result)
219
220    def visual_search_engine(self):
221        return self._visual_search_engine(subdomain="vesa")
222
223
224class CheckerKnowledgeHub(CheckerBase):
225    """
226    Checker for the Knowledgehub Service
227    """
228
229    checker_config = CheckerConfigs("knowledgehub")
230
231    @resolve_url
232    def _sparql(self):
233        """
234        Checks if the Knowledgehub-SPARQL Endpoint is availiable
235        This is done by checking if an anchor of class 'navbar-brand' exists
236        and its text is 'SPARQL Query Editor'
237        """
238        result = self.selenium.find_by_css(css="a[class='navbar-brand']")
239        return getattr(result, "text").strip() == "SPARQL Query Editor"
240
241    def sparql(self):
242        return self._sparql(subdomain="sparql")
243
244    @resolve_url
245    def _cordra(self):
246        """
247        Checks if cordra is available,
248        This is done by checking for the searchbar.
249        """
250        result = self.selenium.find_by_css(css="input[placeholder='Search']")
251
252        return bool(result)
253
254    def cordra(self):
255        return self._cordra(subdomain="cordra")
256
257    @resolve_url
258    def landing_page_knowledgehub(self):
259        """
260        Checks if the Landingpage of Knowlegehub is available.
261        """
262        current_url = self.selenium.get_url()
263        result = current_url.startswith("https://knowledgehub")
264
265        return bool(result)
266
267    @resolve_url
268    def more_examples_loaded(self):
269        """
270        Check if more cards appear after button click.
271        """
272        toggle_button = self.selenium.find_by_css(
273            css=("#hidden-card-deck-toggle")
274        )
275        if toggle_button is None:
276            log.warning(f"{toggle_button} not found")
277            return False
278        else:
279            self.selenium.zoom()
280            toggle_button.click()
281            result = self.selenium.wait_until(
282                css="div[class='row mb-3'][id='card-deck']", want_bool=True
283            )
284        return result
285
286    @resolve_url
287    def table_is_displayed(self):
288        """
289        Checks if the "Harvested Ressources table appears
290        with atleast 10 Entries fetched.
291        """
292        self.selenium.zoom()
293        self.selenium.save_screenshot("/tmp/table.png")
294        self.selenium.wait_until(css=".col-3.border")
295        result = self.selenium.amount_of_elements(css=".col-3.border")
296        return bool(result)
297
298
299class CheckerSupportApps(CheckerBase):
300    """
301    Checker for the SupportApps Service
302    """
303
304    checker_config = CheckerConfigs("supportapps")
305
306    @resolve_url
307    def _jupyterhub(self):
308        """
309        Checks if entry level of JupyterHub is restricted / service is running
310        This is done by checking if Logo and corresponding error code is shown.
311        """
312
313        results = []
314
315        #  Check error
316        result = self.selenium.find_by_text(
317            css=".error",
318            text="405 : Method Not Allowed",
319            case_sensitive=False,
320        )
321        results.append(result)
322
323        #  Check logo
324        result = self.selenium.find_by_css(css="img[src='/hub/logo']")
325        results.append(result)
326
327        return all(results)
328
329    def jupyterhub(self):
330        return self._jupyterhub(subdomain="jupyterhub")
331
332
333class CheckerEduTrain(CheckerBase):
334    """
335    Checker for the EduTrain Service
336    """
337
338    checker_config = CheckerConfigs("edutrain")
339
340    @resolve_url
341    def openedx_lms(self):
342        """
343        Check if the main page is loaded correctly.
344        This is done by checking if the DFG logo is present.
345        """
346        result = self.selenium.find_by_css(
347            # Path may change on a regular, check again once more
348            css="a.explore-btn[href='/courses']"
349        )
350
351        return bool(result)
352
353    @resolve_url
354    def _openedx_cms(self):
355        """
356        Check if Educational Portal is Working
357        This is achieved by checking for the appearance of the logo
358        """
359        result = self.selenium.find_by_css(css="a.brand-link img.brand-image")
360        return bool(result)
361
362    def openedx_cms(self):
363        return self._openedx_cms(subdomain="teacher")
364
365    @resolve_url
366    def _authentication(self):
367        """
368        Check if URL is available, if "AcademicID" button
369        is available and leads to login page
370        """
371        results = []
372        academy_button = self.selenium.find_by_css(
373            css='button[id="oa2-academicid-oauth2"]'
374        )
375
376        academy_button.click()
377
378        current_url = self.selenium.get_url()
379        self.selenium.save_screenshot("/tmp/auth.png")
380        target_page_reached = current_url.startswith(
381            "https://sso.academiccloud.de/simplesaml/module.php/"
382        )
383
384        results.append(target_page_reached)
385        return all(results)
386
387    def authentication(self):
388        return self._authentication(
389            subdomain="apps", path="/authn/login?next=%2F"
390        )
391
392
393class CheckerOneStop4All(CheckerBase):
394    """
395    Checker for the OneStop4All Service
396    """
397
398    checker_config = CheckerConfigs("onestop4all")
399
400    def __init__(self, url=None, instance=None, *args, **kwargs):
401        super().__init__(url=url, instance=instance, *args, **kwargs)
402        self.shadow_root = "onestop4all-app"
403
404    @resolve_url
405    def main_page(self):
406        """
407        Check if the main page is loaded correctly.
408        This is done by checking if the DFG logo is present.
409        """
410        result = self.selenium.find_by_css(
411            css='img[src="/dfg-logo.svg"]', elem=self.shadow_host
412        )
413        return bool(result)
414
415    @resolve_url
416    def number_of_entities(self):
417        """
418        Check if the number of entities is correct.
419        By checking if there is the correct amount of buttons on the webpage.
420        """
421        results = self.selenium.find_by_css(
422            css="div.resourceEntries",
423            elem=self.shadow_host,
424            as_list=True,
425        )
426        return len(results) == O4A_ENTITIES_COUNT
427
428    @resolve_url
429    def involvement(self):
430        """
431        Check if the buttons withing involvement section is loaded correctly.
432        This is simply done, by checking if there are any buttons.
433        """
434        results = self.selenium.find_by_css(
435            css="div.get-involved div.overlap > img",
436            elem=self.shadow_host,
437            as_list=True,
438        )
439        return len(results) > 0
440
441    @resolve_url
442    def support_form(self, close_modal=True):
443        """
444        Check if the support form is loaded correctly.
445        This is done by checking if the subject input field is present.
446        """
447        # try to find the support form button
448        support_form_button = self.selenium.find_by_text(
449            css=".chakra-link",
450            text="user Support",
451            elem=self.shadow_host,
452            case_sensitive=False,
453        )
454
455        # stop check if no support form button was found
456        if not support_form_button:
457            return False
458
459        # click on the support form button to open modal window
460        support_form_button.click()
461
462        subject = self.selenium.find_by_css(
463            css="input[name='subject']", elem=self.shadow_host
464        )
465        if subject and close_modal:
466            # close the modal window finally
467            self.selenium.find_by_css(
468                css="button[aria-label='Close']", elem=self.shadow_host
469            ).click()
470        return bool(subject)
471
472    @resolve_url
473    def captcha_in_support_form(self):
474        """
475        Check if the CAPTCHA is loaded correctly in the support form.
476        This is done by:
477            1. checking if the CAPTCHA checkbox is present
478            2. clicking on the CAPTCHA checkbox
479            3. checking if the CAPTCHA input field is present AND
480                its value is not empty
481        """
482        if self.support_form(close_modal=False):
483            altcha_checkbox = self.selenium.wait_until(
484                css="input#altcha_checkbox",
485                elem=self.shadow_host,
486            )
487            if altcha_checkbox:
488                altcha_checkbox.click()
489                altcha_input = self.selenium.wait_until(
490                    css="input[type='hidden'][name='altcha']",
491                    elem=self.shadow_host,
492                )
493                if altcha_input and altcha_input.get_attribute("value") != "":
494                    return True
495        return False
496
497    @resolve_url
498    def _solr(self):
499        """
500        Check if the Solr login page is loaded correctly.
501        This is done by checking if the Solr logo is present.
502        """
503        login = self.selenium.wait_until(css="button.btn-danger")
504        if login:
505            return login is not None and login.text == "Login"
506
507    def solr(self):
508        return self._solr(path="/solr/#/", ignore_shadow_root=True)