python/samba/tests/blackbox/mdsearch.py

   1 #
   2 # Blackbox tests for mdsearch
   3 #
   4 # Copyright (C) Ralph Boehme                    2019
   5 #
   6 # This program is free software; you can redistribute it and/or modify
   7 # it under the terms of the GNU General Public License as published by
   8 # the Free Software Foundation; either version 3 of the License, or
   9 # (at your option) any later version.
  10 #
  11 # This program is distributed in the hope that it will be useful,
  12 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14 # GNU General Public License for more details.
  15 #
  16 # You should have received a copy of the GNU General Public License
  17 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
  18 #
  19
  20 """Blackbox test for mdsearch"""
  21
  22 import os
  23 import time
  24 import threading
  25 import logging
  26 import json
  27 from http.server import HTTPServer, BaseHTTPRequestHandler
  28 from samba.dcerpc import mdssvc
  29 from samba.tests import BlackboxTestCase
  30 from samba.samba3 import mdscli
  31 from samba.logger import get_samba_logger
  32
  33 logger = get_samba_logger(name=__name__)
  34
  35 testfiles = [
  36     "foo",
  37     "bar",
  38 ]
  39
  40 class MdssvcHTTPRequestHandler(BaseHTTPRequestHandler):
  41     def do_POST(self):
  42         content_length = int(self.headers['content-length'])
  43         body = self.rfile.read(content_length)
  44
  45         actual_json = json.loads((body))
  46         expected_json = json.loads(self.server.json_in)
  47
  48         if actual_json != expected_json:
  49             logger.error("Bad request, expected:\n%s\nGot:\n%s\n" % (expected_json, actual_json))
  50             self.send_error(400,
  51                             "Bad request",
  52                             "Expected: %s\n"
  53                             "Got: %s\n" %
  54                             (expected_json, actual_json))
  55             return
  56
  57         resp = bytes(self.server.json_out, encoding="utf-8")
  58
  59         self.send_response(200)
  60         self.send_header('content-type', 'application/json; charset=UTF-8')
  61         self.send_header('content-length', len(resp))
  62         self.end_headers()
  63         self.wfile.write(resp)
  64
  65 class MdfindBlackboxTests(BlackboxTestCase):
  66
  67     def setUp(self):
  68         super().setUp()
  69
  70         self.server = HTTPServer(('10.53.57.35', 8080),
  71                                  MdssvcHTTPRequestHandler,
  72                                  bind_and_activate=False)
  73
  74         self.t = threading.Thread(target=MdfindBlackboxTests.http_server, args=(self,))
  75         self.t.setDaemon(True)
  76         self.t.start()
  77         time.sleep(1)
  78
  79         self.sharepath = os.environ["LOCAL_PATH"]
  80
  81         for file in testfiles:
  82             f = open("%s/%s" % (self.sharepath, file), "w")
  83             f.close()
  84
  85     def tearDown(self):
  86         super(BlackboxTestCase, self).tearDown()
  87         for file in testfiles:
  88             os.remove("%s/%s" % (self.sharepath, file))
  89
  90     def http_server(self):
  91         self.server.server_bind()
  92         self.server.server_activate()
  93         self.server.serve_forever()
  94
  95     def test_mdsearch(self):
  96         """Simple blackbox test for mdsearch"""
  97
  98         username = os.environ["USERNAME"]
  99         password = os.environ["PASSWORD"]
 100         config = os.environ["SMB_CONF_PATH"]
 101
 102         json_in = r'''{
 103           "from": 0, "size": 50, "_source": ["path.real"],
 104           "query": {
 105             "query_string": {
 106               "query": "(samba*) AND path.real.fulltext:\"%BASEPATH%\""
 107             }
 108           }
 109         }'''
 110         json_out = '''{
 111           "hits" : {
 112             "total" : { "value" : 2},
 113             "hits" : [
 114               {"_source" : {"path" : {"real" : "%BASEPATH%/foo"}}},
 115               {"_source" : {"path" : {"real" : "%BASEPATH%/bar"}}}
 116             ]
 117           }
 118         }'''
 119
 120         self.server.json_in = json_in.replace("%BASEPATH%", self.sharepath)
 121         self.server.json_out = json_out.replace("%BASEPATH%", self.sharepath)
 122
 123         output = self.check_output("mdsearch --configfile=%s -U %s%%%s fileserver spotlight '*==\"samba*\"'" % (config, username, password))
 124
 125         actual = output.decode('utf-8').splitlines()
 126         self.assertEqual(testfiles, actual)