<commit_before># vim:ts=4:sw=4:et:
# Copyright 2012-present Facebook, Inc.
# Licensed under the Apache License, Version 2.0

from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
# no unicode literals

import WatchmanTestCase
import json
import os
import os.path
import sys
import time

WATCHMAN_SRC_DIR = os.environ.get('WATCHMAN_SRC_DIR', os.getcwd())
THIS_DIR = os.path.join(WATCHMAN_SRC_DIR, 'tests', 'integration')


@WatchmanTestCase.expand_matrix
class TestTrigger(WatchmanTestCase.WatchmanTestCase):
    def fileContains(self, file_name, thing):
        if not os.path.exists(file_name):
            return False

        thing = thing + '\n'
        with open(file_name, 'r') as f:
            return thing in f

    def fileHasValidJson(self, file_name):
        if not os.path.exists(file_name):
            return False

        try:
            with open(file_name, 'r') as f:
                json.load(f)
            return True
        except Exception:
            return False

    def checkOSApplicability(self):
        if os.name == 'nt':
            self.skipTest('no append on Windows')

    def test_triggerChdir(self):
        root = self.mkdtemp()
        os.mkdir(os.path.join(root, 'sub'))
        self.watchmanCommand('watch', root)

        self.watchmanCommand('trigger', root, {
            'name': 'cap',
            'command': [
                sys.executable,
                os.path.join(THIS_DIR, 'trig-cwd.py'),
            ],
            'stdout': '>%s' % os.path.join(root, 'trig.log'),
            'expression': ['suffix', 'txt'],
            'stdin': '/dev/null',
            'chdir': 'sub'})

        self.touchRelative(root, 'A.txt')
        self.assertWaitFor(lambda: self.fileContains(
            os.path.join(root, 'trig.log'),
            'PWD=' + os.path.join(root, 'sub')))
        self.assertTrue(
            self.fileContains(
                os.path.join(root, 'trig.log'),
                'WATCHMAN_EMPTY_ENV_VAR='))

    def test_triggerChdirRelativeRoot(self):
        root = self.mkdtemp()
        os.mkdir(os.path.join(root, 'sub1'))
        os.mkdir(os.path.join(root, 'sub1', 'sub2'))
        self.watchmanCommand('watch', root)

        self.watchmanCommand('trigger', root, {
            'name': 'cap',
            'command': [
                sys.executable,
                os.path.join(THIS_DIR, 'trig-cwd.py'),
            ],
            'stdout': '>%s' % os.path.join(root, 'trig.log'),
            'expression': ['suffix', 'txt'],
            'relative_root': 'sub1',
            'stdin': '/dev/null',
            'chdir': 'sub2'})

        self.touchRelative(root, 'sub1', 'A.txt')
        self.assertWaitFor(lambda: self.fileContains(
            os.path.join(root, 'trig.log'),
            'PWD=' + os.path.join(root, 'sub1', 'sub2')))

        self.assertTrue(
            self.fileContains(
                os.path.join(root, 'trig.log'),
                'WATCHMAN_ROOT=' + root))
        self.assertTrue(
            self.fileContains(
                os.path.join(root, 'trig.log'),
                'WATCHMAN_RELATIVE_ROOT=' + os.path.join(root, 'sub1')))

    def test_triggerMaxFiles(self):
        root = self.mkdtemp()<commit_msg>make trigger_chdir less flaky<commit_after># vim:ts=4:sw=4:et:
# Copyright 2012-present Facebook, Inc.
# Licensed under the Apache License, Version 2.0

from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
# no unicode literals

import WatchmanTestCase
import json
import os
import os.path
import sys
import time

WATCHMAN_SRC_DIR = os.environ.get('WATCHMAN_SRC_DIR', os.getcwd())
THIS_DIR = os.path.join(WATCHMAN_SRC_DIR, 'tests', 'integration')


@WatchmanTestCase.expand_matrix
class TestTrigger(WatchmanTestCase.WatchmanTestCase):
    def fileContains(self, file_name, thing):
        if not os.path.exists(file_name):
            return False

        thing = thing + '\n'
        with open(file_name, 'r') as f:
            return thing in f

    def fileHasValidJson(self, file_name):
        if not os.path.exists(file_name):
            return False

        try:
            with open(file_name, 'r') as f:
                json.load(f)
            return True
        except Exception:
            return False

    def checkOSApplicability(self):
        if os.name == 'nt':
            self.skipTest('no append on Windows')

    def test_triggerChdir(self):
        root = self.mkdtemp()
        os.mkdir(os.path.join(root, 'sub'))
        self.watchmanCommand('watch', root)

        self.watchmanCommand('trigger', root, {
            'name': 'cap',
            'command': [
                sys.executable,
                os.path.join(THIS_DIR, 'trig-cwd.py'),
            ],
            'stdout': '>%s' % os.path.join(root, 'trig.log'),
            'expression': ['suffix', 'txt'],
            'stdin': '/dev/null',
            'chdir': 'sub'})

        self.touchRelative(root, 'A.txt')
        self.assertWaitFor(lambda: self.fileContains(
            os.path.join(root, 'trig.log'),
            'PWD=' + os.path.join(root, 'sub')))
        self.assertWaitFor(lambda:
            self.fileContains(
                os.path.join(root, 'trig.log'),
                'WATCHMAN_EMPTY_ENV_VAR='))

    def test_triggerChdirRelativeRoot(self):
        root = self.mkdtemp()
        os.mkdir(os.path.join(root, 'sub1'))
        os.mkdir(os.path.join(root, 'sub1', 'sub2'))
        self.watchmanCommand('watch', root)

        self.watchmanCommand('trigger', root, {
            'name': 'cap',
            'command': [
                sys.executable,
                os.path.join(THIS_DIR, 'trig-cwd.py'),
            ],
            'stdout': '>%s' % os.path.join(root, 'trig.log'),
            'expression': ['suffix', 'txt'],
            'relative_root': 'sub1',
            'stdin': '/dev/null',
            'chdir': 'sub2'})

        self.touchRelative(root, 'sub1', 'A.txt')
        self.assertWaitFor(lambda: self.fileContains(
            os.path.join(root, 'trig.log'),
            'PWD=' + os.path.join(root, 'sub1', 'sub2')))

        self.assertWaitFor(
            lambda: self.fileContains(
                os.path.join(root, 'trig.log'),
                'WATCHMAN_ROOT=' + root))
        self.assertWaitFor(
            lambda: self.fileContains(
                os.path.join(root, 'trig.log'),
                'WATCHMAN_RELATIVE_ROOT=' + os.path.join(root, 'sub1')))

    def test_triggerMaxFiles(self):
        root = self.mkdtemp()
{'fraction_non_alphanumeric': 0.1400121617512922, 'fraction_numerical': 0.005168744299179082, 'mean_word_length': 2.0628491620111733, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2479430', 'n_tokens_mistral': 2112, 'n_tokens_neox': 2032, 'n_words': 408}