<commit_before>import json
import subprocess
import struct
import sys
import unittest
import testconf

# The protocol datagram is described here:
# https://developer.chrome.com/extensions/nativeMessaging#native-messaging-host-protocol

class TestHostPipe(unittest.TestCase):

  def get_response(self):
      response_length = struct.unpack("=I", self.p.stdout.read(4))[0]
      response = str(self.p.stdout.read(response_length))
      # make it into "oneline" json before printing
      response_print = json.dumps(json.loads(response))
      print ("RECV: %s" % response_print)
      return json.loads(response)

  def transceive(self, msg):
      # send like described in 
      print ("SEND: %s" % msg)
      self.p.stdin.write(struct.pack("=I", len(msg)))
      self.p.stdin.write(msg)
      # now read the input
      return self.get_response()

  def setUp(self):
      should_close_fds = sys.platform.startswith('win32') == False;
      self.p = subprocess.Popen(testconf.get_exe(), stdin=subprocess.PIPE, stdout=subprocess.PIPE, close_fds=should_close_fds, stderr=None)
      print ("Running native component on PID %d" % self.p.pid)

  def tearDown(self):
      self.p.terminate()
      self.p.wait()

  def test_random_string(self):
      cmd = "BLAH"
      resp = self.transceive(cmd)
      self.assertEquals(resp["result"], "invalid_argument")

  def test_plain_string(self):
      self.p.stdin.write("Hello World!")
      resp = self.get_response()
      self.assertEquals(resp["result"], "invalid_argument")

  def test_utopic_length(self):
      # write big bumber and little data
      self.p.stdin.write(struct.pack("=I", 0xFFFFFFFF))
      self.p.stdin.write("Hello World!")
      resp = self.get_response()
      self.assertEquals(resp["result"], "invalid_argument")
  
if __name__ == '__main__':
    # run tests
    unittest.main()<commit_msg>Add all headless tests to automatic tests against single instance<commit_after>import json
import subprocess
import struct
import sys
import unittest
import testconf
import uuid

# The protocol datagram is described here:
# https://developer.chrome.com/extensions/nativeMessaging#native-messaging-host-protocol

class TestHostPipe(unittest.TestCase):

  def get_response(self):
      response_length = struct.unpack("=I", self.p.stdout.read(4))[0]
      response = str(self.p.stdout.read(response_length))
      # make it into "oneline" json before printing
      response_print = json.dumps(json.loads(response))
      print("RECV: %s" % response_print)
      return json.loads(response)

  def transceive(self, msg):
      # send like described in 
      print("SEND: %s" % msg)
      self.p.stdin.write(struct.pack("=I", len(msg)))
      self.p.stdin.write(msg)
      # now read the input
      return self.get_response()

  def setUp(self):
      should_close_fds = sys.platform.startswith('win32') == False;
      self.p = subprocess.Popen(testconf.get_exe(), stdin=subprocess.PIPE, stdout=subprocess.PIPE, close_fds=should_close_fds, stderr=None)
      print("\nRunning native component on PID %d" % self.p.pid)

  def tearDown(self):
      self.p.terminate()
      self.p.wait()

  def test_random_string(self):
      cmd = "BLAH"
      resp = self.transceive(cmd)
      self.assertEquals(resp["result"], "invalid_argument")

  def test_plain_string(self):
      self.p.stdin.write("Hello World!")
      resp = self.get_response()
      self.assertEquals(resp["result"], "invalid_argument")

  def test_utopic_length(self):
      # write big bumber and little data
      self.p.stdin.write(struct.pack("=I", 0xFFFFFFFF))
      self.p.stdin.write("Hello World!")
      resp = self.get_response()
      self.assertEquals(resp["result"], "invalid_argument")
  
  # other headless tests
  def test_version_no_nonce(self):
      cmd = {"type": "VERSION", "origin": "https://example.com/"}
      resp = self.transceive(json.dumps(cmd))
      self.assertEqual(resp["result"], "invalid_argument")

  def test_version_no_origin(self):
      cmd = {"type": "VERSION", "nonce": str(uuid.uuid4())}
      resp = self.transceive(json.dumps(cmd))
      self.assertEqual(resp["result"], "invalid_argument")

  def test_version_invalid_origin(self):
      cmd = {"type": "VERSION", "nonce": str(uuid.uuid4()), "origin": "foobar in da house"}
      resp = self.transceive(json.dumps(cmd))
      self.assertEqual(resp["result"], "not_allowed")

  def test_version_file_origin(self):
      cmd = {"type": "VERSION", "nonce": str(uuid.uuid4()), "origin": "file:///tmp/index.html"}
      resp = self.transceive(json.dumps(cmd))
      self.assertEqual(resp["result"], "ok")
      self.assertTrue(re.compile("^\d\.\d+\.\d{1,3}$").match(resp["version"]))

  def test_version_http_origin(self):
      cmd = {"type": "VERSION", "nonce": str(uuid.uuid4()), "origin": "http://example.com/"}
      resp = self.transceive(json.dumps(cmd))
      self.assertEqual(resp["result"], "ok")
      self.assertTrue(re.compile("^\d\.\d+\.\d{1,3}$").match(resp["version"]))

  def test_version_https(self):
      cmd = {"type": "VERSION", "nonce": str(uuid.uuid4()), "origin": "https://example.com/"}
      resp = self.transceive(json.dumps(cmd))
      self.assertEqual(resp["result"], "ok")
      self.assertTrue(re.compile("^\d\.\d+\.\d{1,3}$").match(resp["version"]))


if __name__ == '__main__':
    # run tests
    unittest.main()
{'fraction_non_alphanumeric': 0.15622669649515286, 'fraction_numerical': 0.0039149888143176735, 'mean_word_length': 3.735216240070609, 'pattern_counts': {'":': 16, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 4, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 2, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6575120', 'n_tokens_mistral': 1782, 'n_tokens_neox': 1754, 'n_words': 338}