<commit_before>/**
    Copyright 2013 <NAME>

   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
   You may obtain a copy of the License at

       http://www.apache.org/licenses/LICENSE-2.0

   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.
 */
package net.sf.xenqtt.mockbroker;

import static org.junit.Assert.*;

import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

import net.sf.xenqtt.ApplicationArguments;
import net.sf.xenqtt.MqttTimeoutException;
import net.sf.xenqtt.client.MqttClient;
import net.sf.xenqtt.client.MqttClientListener;
import net.sf.xenqtt.client.NullReconnectStrategy;
import net.sf.xenqtt.client.PublishMessage;
import net.sf.xenqtt.client.Subscription;
import net.sf.xenqtt.client.SynchronousMqttClient;
import net.sf.xenqtt.message.ConnectReturnCode;
import net.sf.xenqtt.message.QoS;

import org.junit.Test;

public class MockBrokerApplicationTest {

	MockBrokerApplication application = new MockBrokerApplication();

	@Test
	public void testStart_NotAuthorized() {
		Map<String, String> args = new HashMap<String, String>();
		args.put("-p", "0");
		ApplicationArguments arguments = new ApplicationArguments(new ArrayList<String>(), args);
		application.start(arguments);

		int port = getPort();
		MqttClientListener listener = new MqttClientListener() {

			@Override
			public void publishReceived(MqttClient client, PublishMessage message) {
			}

			@Override
			public void disconnected(MqttClient client, Throwable cause, boolean reconnecting) {
			}

		};
		SynchronousMqttClient client = new SynchronousMqttClient(String.format("tcp://localhost:%d", port), listener, new NullReconnectStrategy(), 1, 1, 5, 1);
		assertSame(ConnectReturnCode.NOT_AUTHORIZED, client.connect("clientId", true, 15));
	}

	@Test
	public void testStart_NotAuthorized_BadCredentials() throws Exception {
		Map<String, String> args = new HashMap<String, String>();
		args.put("-p", "0");
		args.put("-u", "user1;pass1");
		ApplicationArguments arguments = new ApplicationArguments(new ArrayList<String>(), args);
		application.start(arguments);

		int port = getPort();
		MqttClientListener listener = new MqttClientListener() {

			@Override
			public void publishReceived(MqttClient client, PublishMessage message) {
			}

			@Override
			public void disconnected(MqttClient client, Throwable cause, boolean reconnecting) {
			}

		};
		SynchronousMqttClient client = new SynchronousMqttClient(String.format("tcp://localhost:%d", port), listener, new NullReconnectStrategy(), 1, 1, 5, 1);
		assertSame(ConnectReturnCode.BAD_CREDENTIALS, client.connect("clientId", true, 15, "user1", "pass2"));
	}

	@Test
	public void testStart_AnonymousAuthorization() throws Exception {
		List<String> flags = new ArrayList<String>();
		flags.add("-a");
		Map<String, String> args = new HashMap<String, String>();
		args.put("-p", "0");
		ApplicationArguments arguments = new ApplicationArguments(flags, args);
		application.start(arguments);

		int port = getPort();
		final List<String> messagePayloads = new ArrayList<String>();
		final CountDownLatch latch = new CountDownLatch(1);
		MqttClientListener listener = new MqttClientListener() {

			@Override
			public void publishReceived(MqttClient client, PublishMessage message) {
				messagePayloads.add(message.getPayloadString());
				message.ack();
				latch.countDown();
			}

			@Override
			public void disconnected(MqttClient client, Throwable cause, boolean reconnecting) {
			}

		};
		SynchronousMqttClient client = new SynchronousMqttClient(String.format("tcp://localhost:%d", port), listener, new NullReconnectStrategy(), 1, 0, 5, 0);
		assertSame(ConnectReturnCode.ACCEPTED, client.connect("clientId", true, 15));

		Subscription[] subscriptions = new Subscription[] { new Subscription("grand/foo/bar", QoS.AT_LEAST_ONCE) };
		assertArrayEquals(subscriptions, client.subscribe(subscriptions));
		client.publish(new PublishMessage("grand/foo/bar", QoS.AT_LEAST_ONCE, "onyx"));
		assertTrue(latch.await(1, TimeUnit.SECONDS));
		assertEquals(1, messagePayloads.size());
		assertEquals("onyx", messagePayloads.get(0));
	}

	@Test
	public void testStart_CredentialAuthorization() throws Exception {
		Map<String, String> args = new HashMap<String, String>();
		args.put("-p", "0");
		args.put("-u", "user1;pass1");
		ApplicationArguments arguments = new ApplicationArguments(new ArrayList<String>(), args);
		application.start(arguments);

		int port = getPort();
		final List<String> messagePayloads = new ArrayList<String>();
		final CountDownLatch latch = new CountDownLatch(1);
		MqttClientListener listener = new MqttClientListener() {

			@Override
			public void publishReceived(MqttClient client, PublishMessage message) {
				messagePayloads.add(message.getPayloadString());
				message.ack();
				latch.countDown();
			}

			@Override
			public void disconnected(MqttClient client, Throwable cause, boolean reconnecting) {
			}

		};
		SynchronousMqttClient client = new SynchronousMqttClient(String.format("tcp://localhost:%d", port), listener, new NullReconnectStrategy(), 1, 0, 5, 0);
		assertSame(ConnectReturnCode.ACCEPTED, client.connect("clientId", true, 15, "user1", "<PASSWORD>"));

		Subscription[] subscriptions = new Subscription[] { new Subscription("grand/foo/bar", QoS.AT_LEAST_ONCE) };
		assertArrayEquals(subscriptions, client.subscribe(subscriptions));
		client.publish(new PublishMessage("grand/foo/bar", QoS.AT_LEAST_ONCE, "onyx"));
		assertTrue(latch.await(1, TimeUnit.SECONDS));
		assertEquals(1, messagePayloads.size());
		assertEquals("onyx", messagePayloads.get(0));
	}

	@Test
	public void testStart_NonDefaultResendInterval() throws Exception {
		Map<String, String> args = new HashMap<String, String>();
		args.put("-p", "0");
		args.put("-t", "2");
		args.put("-u", "user1;pass1");
		ApplicationArguments arguments = new ApplicationArguments(new ArrayList<String>(), args);
		application.start(arguments);

		int port = getPort();
		final List<String> messagePayloads = new ArrayList<String>();
		final CountDownLatch latch = new CountDownLatch(1);
		final AtomicLong firstReceived = new AtomicLong();
		MqttClientListener listener = new MqttClientListener() {

			@Override
			public void publishReceived(MqttClient client, PublishMessage message) {
				long now = System.currentTimeMillis();
				if (firstReceived.get() == 0) {
					firstReceived.set(now);
					return;
				}

				messagePayloads.add(message.getPayloadString());
				message.ack();
				latch.countDown();<commit_msg>extended latch wait timeout in a couple tests<commit_after>/**
    Copyright 2013 <NAME>

   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
   You may obtain a copy of the License at

       http://www.apache.org/licenses/LICENSE-2.0

   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.
 */
package net.sf.xenqtt.mockbroker;

import static org.junit.Assert.*;

import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

import net.sf.xenqtt.ApplicationArguments;
import net.sf.xenqtt.MqttTimeoutException;
import net.sf.xenqtt.client.MqttClient;
import net.sf.xenqtt.client.MqttClientListener;
import net.sf.xenqtt.client.NullReconnectStrategy;
import net.sf.xenqtt.client.PublishMessage;
import net.sf.xenqtt.client.Subscription;
import net.sf.xenqtt.client.SynchronousMqttClient;
import net.sf.xenqtt.message.ConnectReturnCode;
import net.sf.xenqtt.message.QoS;

import org.junit.Test;

public class MockBrokerApplicationTest {

	MockBrokerApplication application = new MockBrokerApplication();

	@Test
	public void testStart_NotAuthorized() {
		Map<String, String> args = new HashMap<String, String>();
		args.put("-p", "0");
		ApplicationArguments arguments = new ApplicationArguments(new ArrayList<String>(), args);
		application.start(arguments);

		int port = getPort();
		MqttClientListener listener = new MqttClientListener() {

			@Override
			public void publishReceived(MqttClient client, PublishMessage message) {
			}

			@Override
			public void disconnected(MqttClient client, Throwable cause, boolean reconnecting) {
			}

		};
		SynchronousMqttClient client = new SynchronousMqttClient(String.format("tcp://localhost:%d", port), listener, new NullReconnectStrategy(), 1, 1, 5, 1);
		assertSame(ConnectReturnCode.NOT_AUTHORIZED, client.connect("clientId", true, 15));
	}

	@Test
	public void testStart_NotAuthorized_BadCredentials() throws Exception {
		Map<String, String> args = new HashMap<String, String>();
		args.put("-p", "0");
		args.put("-u", "user1;<PASSWORD>");
		ApplicationArguments arguments = new ApplicationArguments(new ArrayList<String>(), args);
		application.start(arguments);

		int port = getPort();
		MqttClientListener listener = new MqttClientListener() {

			@Override
			public void publishReceived(MqttClient client, PublishMessage message) {
			}

			@Override
			public void disconnected(MqttClient client, Throwable cause, boolean reconnecting) {
			}

		};
		SynchronousMqttClient client = new SynchronousMqttClient(String.format("tcp://localhost:%d", port), listener, new NullReconnectStrategy(), 1, 1, 5, 1);
		assertSame(ConnectReturnCode.BAD_CREDENTIALS, client.connect("clientId", true, 15, "user1", "<PASSWORD>"));
	}

	@Test
	public void testStart_AnonymousAuthorization() throws Exception {
		List<String> flags = new ArrayList<String>();
		flags.add("-a");
		Map<String, String> args = new HashMap<String, String>();
		args.put("-p", "0");
		ApplicationArguments arguments = new ApplicationArguments(flags, args);
		application.start(arguments);

		int port = getPort();
		final List<String> messagePayloads = new ArrayList<String>();
		final CountDownLatch latch = new CountDownLatch(1);
		MqttClientListener listener = new MqttClientListener() {

			@Override
			public void publishReceived(MqttClient client, PublishMessage message) {
				messagePayloads.add(message.getPayloadString());
				message.ack();
				latch.countDown();
			}

			@Override
			public void disconnected(MqttClient client, Throwable cause, boolean reconnecting) {
			}

		};
		SynchronousMqttClient client = new SynchronousMqttClient(String.format("tcp://localhost:%d", port), listener, new NullReconnectStrategy(), 1, 0, 5, 0);
		assertSame(ConnectReturnCode.ACCEPTED, client.connect("clientId", true, 15));

		Subscription[] subscriptions = new Subscription[] { new Subscription("grand/foo/bar", QoS.AT_LEAST_ONCE) };
		assertArrayEquals(subscriptions, client.subscribe(subscriptions));
		client.publish(new PublishMessage("grand/foo/bar", QoS.AT_LEAST_ONCE, "onyx"));
		assertTrue(latch.await(5, TimeUnit.SECONDS));
		assertEquals(1, messagePayloads.size());
		assertEquals("onyx", messagePayloads.get(0));
	}

	@Test
	public void testStart_CredentialAuthorization() throws Exception {
		Map<String, String> args = new HashMap<String, String>();
		args.put("-p", "0");
		args.put("-u", "user1;pass1");
		ApplicationArguments arguments = new ApplicationArguments(new ArrayList<String>(), args);
		application.start(arguments);

		int port = getPort();
		final List<String> messagePayloads = new ArrayList<String>();
		final CountDownLatch latch = new CountDownLatch(1);
		MqttClientListener listener = new MqttClientListener() {

			@Override
			public void publishReceived(MqttClient client, PublishMessage message) {
				messagePayloads.add(message.getPayloadString());
				message.ack();
				latch.countDown();
			}

			@Override
			public void disconnected(MqttClient client, Throwable cause, boolean reconnecting) {
			}

		};
		SynchronousMqttClient client = new SynchronousMqttClient(String.format("tcp://localhost:%d", port), listener, new NullReconnectStrategy(), 1, 0, 5, 0);
		assertSame(ConnectReturnCode.ACCEPTED, client.connect("clientId", true, 15, "user1", "pass1"));

		Subscription[] subscriptions = new Subscription[] { new Subscription("grand/foo/bar", QoS.AT_LEAST_ONCE) };
		assertArrayEquals(subscriptions, client.subscribe(subscriptions));
		client.publish(new PublishMessage("grand/foo/bar", QoS.AT_LEAST_ONCE, "onyx"));
		assertTrue(latch.await(5, TimeUnit.SECONDS));
		assertEquals(1, messagePayloads.size());
		assertEquals("onyx", messagePayloads.get(0));
	}

	@Test
	public void testStart_NonDefaultResendInterval() throws Exception {
		Map<String, String> args = new HashMap<String, String>();
		args.put("-p", "0");
		args.put("-t", "2");
		args.put("-u", "user1;<PASSWORD>");
		ApplicationArguments arguments = new ApplicationArguments(new ArrayList<String>(), args);
		application.start(arguments);

		int port = getPort();
		final List<String> messagePayloads = new ArrayList<String>();
		final CountDownLatch latch = new CountDownLatch(1);
		final AtomicLong firstReceived = new AtomicLong();
		MqttClientListener listener = new MqttClientListener() {

			@Override
			public void publishReceived(MqttClient client, PublishMessage message) {
				long now = System.currentTimeMillis();
				if (firstReceived.get() == 0) {
					firstReceived.set(now);
					return;
				}

				messagePayloads.add(message.getPayloadString());
				message.ack();
				latch.countDown();
{'fraction_non_alphanumeric': 0.12162353273935794, 'fraction_numerical': 0.00791967189930703, 'mean_word_length': 5.342152466367713, 'pattern_counts': {'":': 0, '<': 53, '<?xml version=': 0, '>': 53, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, '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': '4671109', 'n_tokens_mistral': 4638, 'n_tokens_neox': 4221, 'n_words': 1028}